From 16676bca49d09ef6255e59859f1c612f649fb709 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sun, 8 Mar 2026 16:37:28 +0100 Subject: [PATCH] perf: watch together CPU overhead --- lib/services/base_peer_service.dart | 9 +++ lib/utils/app_logger.dart | 30 +++++----- .../providers/watch_together_provider.dart | 7 ++- .../services/watch_together_peer_service.dart | 3 - .../services/watch_together_sync_manager.dart | 58 ++++++------------- 5 files changed, 44 insertions(+), 63 deletions(-) diff --git a/lib/services/base_peer_service.dart b/lib/services/base_peer_service.dart index 8d4ae792..c953bfa5 100644 --- a/lib/services/base_peer_service.dart +++ b/lib/services/base_peer_service.dart @@ -36,6 +36,7 @@ class PeerError { mixin KeepaliveMixin { Timer? _pingTimer; Timer? _pongTimer; + DateTime? _lastPongReset; /// How often to send a keepalive ping. Duration get pingInterval; @@ -55,12 +56,19 @@ mixin KeepaliveMixin { void startKeepalive() { _pingTimer?.cancel(); _pingTimer = Timer.periodic(pingInterval, (_) => sendPing()); + _lastPongReset = null; // Force first reset resetPongTimer(); } /// Reset the pong timeout (call on every incoming message). + /// Coalesced: skips if last reset was <5s ago (timeout is 30s, still safe). void resetPongTimer() { if (pongTimeout == Duration.zero) return; + final now = DateTime.now(); + if (_lastPongReset != null && now.difference(_lastPongReset!) < const Duration(seconds: 5)) { + return; + } + _lastPongReset = now; _pongTimer?.cancel(); _pongTimer = Timer(pongTimeout, onPongTimeout); } @@ -71,5 +79,6 @@ mixin KeepaliveMixin { _pingTimer = null; _pongTimer?.cancel(); _pongTimer = null; + _lastPongReset = null; } } diff --git a/lib/utils/app_logger.dart b/lib/utils/app_logger.dart index 46cd1b73..3e6db52d 100644 --- a/lib/utils/app_logger.dart +++ b/lib/utils/app_logger.dart @@ -1,3 +1,5 @@ +import 'dart:collection'; + import 'package:logger/logger.dart'; import 'log_redaction_manager.dart'; @@ -52,13 +54,18 @@ class LogEntry { } /// Custom log output that stores logs in memory with a circular buffer +/// +/// Storage is handled by [MemoryAwareLogPrinter.log()] — this class only +/// forwards formatted lines to the console via the default [ConsoleOutput]. class MemoryLogOutput extends LogOutput { static const int maxLogSizeBytes = 5 * 1024 * 1024; // 5 MB - static final List _logs = []; + static final ListQueue _logs = ListQueue(); static int _currentSize = 0; + static final _consoleOutput = ConsoleOutput(); + /// Get all stored logs (newest first) - static List getLogs() => List.unmodifiable(_logs.reversed); + static List getLogs() => _logs.toList().reversed.toList(); /// Clear all stored logs static void clearLogs() { @@ -74,19 +81,8 @@ class MemoryLogOutput extends LogOutput { @override void output(OutputEvent event) { - // Extract relevant information from the log event - for (var line in event.lines) { - final logEntry = LogEntry(timestamp: DateTime.now(), level: event.level, message: _redactSensitiveData(line)); - - _logs.add(logEntry); - _currentSize += logEntry.estimatedSize; - - // Maintain buffer size limit (remove oldest entries) - while (_currentSize > maxLogSizeBytes && _logs.isNotEmpty) { - final removed = _logs.removeAt(0); - _currentSize -= removed.estimatedSize; - } - } + // Only print to console — storage is done in MemoryAwareLogPrinter.log() + _consoleOutput.output(event); } } @@ -113,9 +109,9 @@ class MemoryAwareLogPrinter extends LogPrinter { MemoryLogOutput._logs.add(logEntry); MemoryLogOutput._currentSize += logEntry.estimatedSize; - // Maintain buffer size limit (remove oldest entries) + // Maintain buffer size limit (remove oldest entries) — O(1) with ListQueue while (MemoryLogOutput._currentSize > MemoryLogOutput.maxLogSizeBytes && MemoryLogOutput._logs.isNotEmpty) { - final removed = MemoryLogOutput._logs.removeAt(0); + final removed = MemoryLogOutput._logs.removeFirst(); MemoryLogOutput._currentSize -= removed.estimatedSize; } diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index deca8f63..ff15a79b 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -440,8 +440,11 @@ class WatchTogetherProvider with ChangeNotifier { if (message.peerId != null) { final index = _participants.indexWhere((p) => p.peerId == message.peerId); if (index >= 0) { - _participants[index] = _participants[index].copyWith(isBuffering: message.bufferingState ?? false); - notifyListeners(); + final newState = message.bufferingState ?? false; + if (_participants[index].isBuffering != newState) { + _participants[index] = _participants[index].copyWith(isBuffering: newState); + notifyListeners(); + } } } break; diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index 076a505c..8f13e788 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -178,7 +178,6 @@ class WatchTogetherPeerService with KeepaliveMixin { try { final payloadStr = payload is String ? payload : jsonEncode(payload); final syncMsg = SyncMessage.fromJson(payloadStr); - appLogger.d('WatchTogether: Received ${syncMsg.type} from $from'); _safeAdd(_messageReceivedController, syncMsg); } catch (e) { appLogger.e('WatchTogether: Failed to parse sync message payload', error: e); @@ -369,14 +368,12 @@ class WatchTogetherPeerService with KeepaliveMixin { /// Broadcast a message to all connected peers void broadcast(SyncMessage message) { final payload = message.toJson(); - appLogger.d('WatchTogether: Broadcasting ${message.type} to ${_connectedPeers.length} peers'); _sendRaw({'type': 'broadcast', 'payload': payload}); } /// Send a message to a specific peer void sendTo(String peerId, SyncMessage message) { final payload = message.toJson(); - appLogger.d('WatchTogether: Sending ${message.type} to $peerId'); _sendRaw({'type': 'sendTo', 'to': peerId, 'payload': payload}); } diff --git a/lib/watch_together/services/watch_together_sync_manager.dart b/lib/watch_together/services/watch_together_sync_manager.dart index cb699a21..686ed71a 100644 --- a/lib/watch_together/services/watch_together_sync_manager.dart +++ b/lib/watch_together/services/watch_together_sync_manager.dart @@ -66,6 +66,9 @@ class WatchTogetherSyncManager { // Timer for clearing sync indicator (prevents flicker from overlapping corrections) Timer? _syncingTimer; + // Debounce timer for buffering broadcasts + Timer? _bufferingDebounceTimer; + // Track last known state to avoid duplicate broadcasts bool _lastKnownPlaying = false; double _lastKnownRate = 1.0; @@ -87,7 +90,6 @@ class WatchTogetherSyncManager { /// Update the session (e.g., when control mode changes) void updateSession(WatchSession session) { _session = session; - appLogger.d('WatchTogether: Sync manager session updated, controlMode: ${session.controlMode}'); } /// Whether this manager has a player attached @@ -172,6 +174,8 @@ class WatchTogetherSyncManager { _firstPlayCompleted = false; _syncingTimer?.cancel(); _syncingTimer = null; + _bufferingDebounceTimer?.cancel(); + _bufferingDebounceTimer = null; _clockSyncTimer?.cancel(); _clockSyncTimer = null; _clockOffset = 0; @@ -246,8 +250,11 @@ class WatchTogetherSyncManager { } } - // Broadcast for UI (peer buffering indicators) — no playback control - _peerService.broadcast(SyncMessage.buffering(isBuffering, peerId: _peerService.myPeerId)); + // Broadcast for UI (peer buffering indicators) — debounced to avoid churn + _bufferingDebounceTimer?.cancel(); + _bufferingDebounceTimer = Timer(const Duration(milliseconds: 300), () { + _peerService.broadcast(SyncMessage.buffering(isBuffering, peerId: _peerService.myPeerId)); + }); }), ); @@ -361,7 +368,6 @@ class WatchTogetherSyncManager { // Exponential moving average const alpha = 0.3; _clockOffset = (_clockOffset + (alpha * (sampleOffset - _clockOffset)).round()); - appLogger.d('WatchTogether: Clock offset updated: ${_clockOffset}ms (sample: ${sampleOffset}ms, RTT: ${rtt}ms)'); } } @@ -504,10 +510,7 @@ class WatchTogetherSyncManager { /// Broadcast play/pause state void _broadcastPlayPause(bool isPlaying) { - if (!_canControl()) { - appLogger.d('WatchTogether: Cannot control playback in hostOnly mode'); - return; - } + if (!_canControl()) return; if (isPlaying) { final position = _player?.state.position ?? Duration.zero; @@ -519,10 +522,7 @@ class WatchTogetherSyncManager { /// Called when user seeks locally void onLocalSeek(Duration position) { - if (!_canControl()) { - appLogger.d('WatchTogether: Cannot control playback in hostOnly mode'); - return; - } + if (!_canControl()) return; _peerService.broadcast(SyncMessage.seek(position, peerId: _peerService.myPeerId)); } @@ -544,7 +544,6 @@ class WatchTogetherSyncManager { message.type == SyncMessageType.rate; if (isControlMessage) { - appLogger.d('WatchTogether: Host relaying ${message.type} from ${message.peerId}'); _peerService.broadcast(message); } } @@ -561,35 +560,23 @@ class WatchTogetherSyncManager { message.type == SyncMessageType.pong || message.type == SyncMessageType.mediaSwitch; - if (!isHostMessage && !isMetaMessage) { - appLogger.d('WatchTogether: Ignoring non-host message in hostOnly mode'); - return; - } + if (!isHostMessage && !isMetaMessage) return; } switch (message.type) { case SyncMessageType.play: - if (!_shouldApplyRemoteControl(message)) { - appLogger.d('WatchTogether: Ignoring play from non-host in hostOnly mode'); - break; - } + if (!_shouldApplyRemoteControl(message)) break; await _applyRemotePlay(position: message.position, expectedAttachmentGeneration: queuedAttachmentGeneration); break; case SyncMessageType.pause: - if (!_shouldApplyRemoteControl(message)) { - appLogger.d('WatchTogether: Ignoring pause from non-host in hostOnly mode'); - break; - } + if (!_shouldApplyRemoteControl(message)) break; _deferredPlay = false; await _applyRemotePause(expectedAttachmentGeneration: queuedAttachmentGeneration); break; case SyncMessageType.seek: - if (!_shouldApplyRemoteControl(message)) { - appLogger.d('WatchTogether: Ignoring seek from non-host in hostOnly mode'); - break; - } + if (!_shouldApplyRemoteControl(message)) break; if (message.position != null) { await _applyRemoteSeek(message.position!, expectedAttachmentGeneration: queuedAttachmentGeneration); } @@ -613,23 +600,18 @@ class WatchTogetherSyncManager { final localPlaying = player.state.playing; if (message.isPlaying! && !localPlaying) { - appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (playing)'); await _applyRemotePlay( position: message.position, expectedAttachmentGeneration: queuedAttachmentGeneration, ); } else if (!message.isPlaying! && localPlaying) { - appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (paused)'); await _applyRemotePause(expectedAttachmentGeneration: queuedAttachmentGeneration); } } break; case SyncMessageType.rate: - if (!_shouldApplyRemoteControl(message)) { - appLogger.d('WatchTogether: Ignoring rate from non-host in hostOnly mode'); - break; - } + if (!_shouldApplyRemoteControl(message)) break; if (message.rate != null) { await _applyRemoteRate(message.rate!, expectedAttachmentGeneration: queuedAttachmentGeneration); } @@ -703,7 +685,6 @@ class WatchTogetherSyncManager { /// Apply remote play command Future _applyRemotePlay({Duration? position, int? expectedAttachmentGeneration}) async { - appLogger.d('WatchTogether: Applying remote PLAY${position != null ? ' at ${position.inSeconds}s' : ''}'); return _runGuardedRemoteAction( actionName: 'play', expectedAttachmentGeneration: expectedAttachmentGeneration, @@ -734,7 +715,6 @@ class WatchTogetherSyncManager { /// Apply remote pause command Future _applyRemotePause({int? expectedAttachmentGeneration}) async { - appLogger.d('WatchTogether: Applying remote PAUSE'); return _runGuardedRemoteAction( actionName: 'pause', expectedAttachmentGeneration: expectedAttachmentGeneration, @@ -755,7 +735,6 @@ class WatchTogetherSyncManager { /// Apply remote seek command Future _applyRemoteSeek(Duration position, {int? expectedAttachmentGeneration}) async { - appLogger.d('WatchTogether: Applying remote SEEK to ${position.inSeconds}s'); return _runGuardedRemoteAction( actionName: 'seek', expectedAttachmentGeneration: expectedAttachmentGeneration, @@ -772,7 +751,6 @@ class WatchTogetherSyncManager { /// Apply remote rate change Future _applyRemoteRate(double rate, {int? expectedAttachmentGeneration}) async { - appLogger.d('WatchTogether: Applying remote RATE: $rate'); return _runGuardedRemoteAction( actionName: 'rate', expectedAttachmentGeneration: expectedAttachmentGeneration, @@ -838,8 +816,6 @@ class WatchTogetherSyncManager { _syncingTimer?.cancel(); _syncingTimer = Timer(const Duration(milliseconds: 500), () => _setSyncing(false)); } else if (drift > maxAllowedDrift) { - // Normal drift correction - appLogger.d('WatchTogether: Drift correction (${drift.inMilliseconds}ms)'); _setSyncing(true); final didSeek = await _applyRemoteSeek( estimatedRemoteNow,