yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: watch together CPU overhead

Browse Source
This commit is contained in:
edde746
2026-03-08 16:57:15 +01:00
parent 9e2e1f72ce
commit 16676bca49
5 changed files with 44 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/services/base_peer_service.dart
+9
View File
@@ -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;
}
}
lib/utils/app_logger.dart
+13 -17
View File
@@ -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<LogEntry> _logs = [];
static final ListQueue<LogEntry> _logs = ListQueue<LogEntry>();
static int _currentSize = 0;
static final _consoleOutput = ConsoleOutput();
/// Get all stored logs (newest first)
static List<LogEntry> getLogs() => List.unmodifiable(_logs.reversed);
static List<LogEntry> 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;
}
lib/watch_together/providers/watch_together_provider.dart
+5 -2
View File
@@ -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;
lib/watch_together/services/watch_together_peer_service.dart
-3
View File
@@ -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});
}
lib/watch_together/services/watch_together_sync_manager.dart
+17 -41
View File
@@ -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<bool> _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<bool> _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<bool> _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<bool> _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,