diff options
Diffstat (limited to 'Emby.Server.Implementations/SyncPlay')
| -rw-r--r-- | Emby.Server.Implementations/SyncPlay/Group.cs | 125 | ||||
| -rw-r--r-- | Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs | 69 |
2 files changed, 189 insertions, 5 deletions
diff --git a/Emby.Server.Implementations/SyncPlay/Group.cs b/Emby.Server.Implementations/SyncPlay/Group.cs index 923bfc67aa..6fbe46ffd6 100644 --- a/Emby.Server.Implementations/SyncPlay/Group.cs +++ b/Emby.Server.Implementations/SyncPlay/Group.cs @@ -11,6 +11,7 @@ using MediaBrowser.Controller.Library; using MediaBrowser.Controller.Session; using MediaBrowser.Controller.SyncPlay; using MediaBrowser.Controller.SyncPlay.GroupStates; +using MediaBrowser.Controller.SyncPlay.PlaybackRequests; using MediaBrowser.Controller.SyncPlay.Queue; using MediaBrowser.Controller.SyncPlay.Requests; using MediaBrowser.Model.SyncPlay; @@ -27,6 +28,11 @@ namespace Emby.Server.Implementations.SyncPlay public class Group : IGroupStateContext { /// <summary> + /// The default value of <see cref="GroupWaitTimeout"/>, in milliseconds. + /// </summary> + internal const long DefaultGroupWaitTimeout = 30000; + + /// <summary> /// The logger. /// </summary> private readonly ILogger<Group> _logger; @@ -54,8 +60,12 @@ namespace Emby.Server.Implementations.SyncPlay /// <summary> /// The participants, or members of the group. /// </summary> - private readonly Dictionary<string, GroupMember> _participants = - new Dictionary<string, GroupMember>(StringComparer.OrdinalIgnoreCase); + private readonly Dictionary<string, GroupMember> _participants = new(StringComparer.OrdinalIgnoreCase); + + /// <summary> + /// The sessions of the participants, which only carry identifiers. + /// </summary> + private readonly Dictionary<string, SessionInfo> _participantSessions = new(StringComparer.OrdinalIgnoreCase); /// <summary> /// The internal group state. @@ -115,6 +125,19 @@ namespace Emby.Server.Implementations.SyncPlay public long MaxPlaybackOffset { get; } = 500; /// <summary> + /// Gets the maximum time, in milliseconds, the group waits for its members to report ready. + /// </summary> + /// <value>The group-wait timeout.</value> + internal long GroupWaitTimeout { get; init; } = DefaultGroupWaitTimeout; + + /// <summary> + /// Gets the <see cref="Environment.TickCount64"/> value at which the group gives up waiting + /// for its members, or <c>null</c> when it is not waiting for anyone. + /// </summary> + /// <value>The group-wait deadline.</value> + internal long? GroupWaitDeadline { get; private set; } + + /// <summary> /// Gets the group identifier. /// </summary> /// <value>The group identifier.</value> @@ -163,6 +186,8 @@ namespace Emby.Server.Implementations.SyncPlay Ping = DefaultPing, IsBuffering = false }); + + _participantSessions[session.Id] = session; } /// <summary> @@ -172,6 +197,8 @@ namespace Emby.Server.Implementations.SyncPlay private void RemoveSession(SessionInfo session) { _participants.Remove(session.Id); + _participantSessions.Remove(session.Id); + UpdateGroupWaitDeadline(false); } /// <summary> @@ -389,13 +416,20 @@ namespace Emby.Server.Implementations.SyncPlay { value.IgnoreGroupWait = ignoreGroupWait; } + + UpdateGroupWaitDeadline(false); } /// <inheritdoc /> public void SetState(IGroupState state) { _logger.LogInformation("Group {GroupId} switching from {FromStateType} to {ToStateType}.", GroupId.ToString(), _state.Type, state.Type); - this._state = state; + _state = state; + + if (state.Type != GroupStateType.Waiting) + { + GroupWaitDeadline = null; + } } /// <inheritdoc /> @@ -475,6 +509,8 @@ namespace Emby.Server.Implementations.SyncPlay { value.IsBuffering = isBuffering; } + + UpdateGroupWaitDeadline(false); } /// <inheritdoc /> @@ -484,6 +520,9 @@ namespace Emby.Server.Implementations.SyncPlay { session.IsBuffering = isBuffering; } + + // Resetting the status of every session starts a new waiting period. + UpdateGroupWaitDeadline(isBuffering); } /// <inheritdoc /> @@ -690,5 +729,85 @@ namespace Emby.Server.Implementations.SyncPlay PlayQueue.ShuffleMode, PlayQueue.RepeatMode); } + + /// <summary> + /// Stops waiting for the members that have not reported ready and lets the rest of the + /// group carry on. Does nothing until <see cref="GroupWaitDeadline"/> has passed. + /// </summary> + /// <param name="cancellationToken">The cancellation token.</param> + internal void HandleGroupWaitTimeout(CancellationToken cancellationToken) + { + var deadline = GroupWaitDeadline; + if (deadline is null || deadline > Environment.TickCount64) + { + return; + } + + GroupWaitDeadline = null; + + if (_state is not WaitingGroupState waitingState) + { + return; + } + + var blockingSessions = _participantSessions + .Values + .Where(participant => _participants.TryGetValue(participant.Id, out var member) + && member.IsBuffering + && !member.IgnoreGroupWait) + .ToList(); + + if (blockingSessions.Count == 0) + { + return; + } + + // The recovery below is broadcast to the whole group, so it does not matter which of + // the sessions that kept the group waiting is the one acting on the group's behalf. + var session = blockingSessions[0]; + + _logger.LogWarning( + "Group {GroupId} waited {Waited} ms for session(s) {SessionIds} to report ready, giving up.", + GroupId.ToString(), + GroupWaitTimeout + Environment.TickCount64 - deadline.Value, + string.Join(", ", blockingSessions.Select(participant => participant.Id))); + + if (waitingState.ResumePlaying) + { + // An unpause request in the waiting state means "start now, ignoring the sessions + // that are not ready". + var unpauseRequest = new UnpauseGroupRequest(); + waitingState.HandleRequest(unpauseRequest, this, GroupStateType.Waiting, session, cancellationToken); + return; + } + + // The members have been paused for the whole waiting period, so the playback position + // stays where the wait started. + SetAllBuffering(false); + SetState(new PausedGroupState(_loggerFactory)); + + var command = NewSyncPlayCommand(SendCommandType.Pause); + SendCommand(session, SyncPlayBroadcastType.AllGroup, command, cancellationToken); + + var stateUpdate = new GroupStateUpdate(GroupStateType.Paused, PlaybackRequestType.Pause); + var update = new SyncPlayStateUpdate(GroupId, stateUpdate); + SendGroupUpdate(session, SyncPlayBroadcastType.AllGroup, update, cancellationToken); + } + + private void UpdateGroupWaitDeadline(bool startNewWaitingPeriod) + { + if (_state.Type != GroupStateType.Waiting || !IsBuffering()) + { + GroupWaitDeadline = null; + return; + } + + // A running deadline covers the waiting period as a whole, so the sessions that keep + // reporting buffering while they load must not push it back. + if (GroupWaitDeadline is null || startNewWaitingPeriod) + { + GroupWaitDeadline = Environment.TickCount64 + GroupWaitTimeout; + } + } } } diff --git a/Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs b/Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs index b88ee33358..88dfb070b8 100644 --- a/Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs +++ b/Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs @@ -19,6 +19,11 @@ namespace Emby.Server.Implementations.SyncPlay public class SyncPlayManager : ISyncPlayManager, IDisposable { /// <summary> + /// How often, in milliseconds, the groups are checked for a spent wait deadline. + /// </summary> + private const int GroupWaitSweepInterval = 1000; + + /// <summary> /// The logger. /// </summary> private readonly ILogger<SyncPlayManager> _logger; @@ -69,6 +74,11 @@ namespace Emby.Server.Implementations.SyncPlay /// </remarks> private readonly Lock _groupsLock = new(); + /// <summary> + /// The timer that watches the groups' wait deadlines, running only while there are groups. + /// </summary> + private readonly Timer _groupWaitTimer; + private bool _disposed = false; /// <summary> @@ -90,8 +100,15 @@ namespace Emby.Server.Implementations.SyncPlay _libraryManager = libraryManager; _logger = loggerFactory.CreateLogger<SyncPlayManager>(); _sessionManager.SessionEnded += OnSessionEnded; + _groupWaitTimer = new Timer(_ => OnGroupWaitTimerTick(), null, Timeout.Infinite, Timeout.Infinite); } + /// <summary> + /// Gets the maximum time, in milliseconds, a group waits for its members to report ready. + /// </summary> + /// <value>The group-wait timeout.</value> + internal long GroupWaitTimeout { get; init; } = Group.DefaultGroupWaitTimeout; + /// <inheritdoc /> public void Dispose() { @@ -122,8 +139,12 @@ namespace Emby.Server.Implementations.SyncPlay LeaveGroup(session, leaveGroupRequest, cancellationToken); } - var group = new Group(_loggerFactory, _userManager, _sessionManager, _libraryManager); + var group = new Group(_loggerFactory, _userManager, _sessionManager, _libraryManager) + { + GroupWaitTimeout = GroupWaitTimeout + }; _groups[group.GroupId] = group; + UpdateGroupWaitTimer(); if (!_sessionToGroupMap.TryAdd(session.Id, group)) { @@ -242,6 +263,7 @@ namespace Emby.Server.Implementations.SyncPlay { _logger.LogInformation("Group {GroupId} is empty, removing it.", group.GroupId); _groups.Remove(group.GroupId, out _); + UpdateGroupWaitTimer(); } } } @@ -384,7 +406,50 @@ namespace Emby.Server.Implementations.SyncPlay } _sessionManager.SessionEnded -= OnSessionEnded; - _disposed = true; + + lock (_groupsLock) + { + _disposed = true; + _groupWaitTimer.Dispose(); + } + } + + private void UpdateGroupWaitTimer() + { + if (_disposed) + { + return; + } + + var interval = _groups.IsEmpty ? Timeout.Infinite : GroupWaitSweepInterval; + _groupWaitTimer.Change(interval, interval); + } + + private void OnGroupWaitTimerTick() + { + try + { + lock (_groupsLock) + { + if (_disposed) + { + return; + } + + foreach (var (_, group) in _groups) + { + // Group lock required as Group is not thread-safe. + lock (group) + { + group.HandleGroupWaitTimeout(CancellationToken.None); + } + } + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Error while recovering SyncPlay groups from a timed out wait."); + } } private void OnSessionEnded(object sender, SessionEventArgs e) |