Merge pull request #13459 from JPVenson/bugfix/13457_FixWebSocketControllerConcurrency

Fixed Websocket not locking state correctly
pull/13499/head
JPVenson 2 weeks ago committed by GitHub
parent 731874429c
commit 49bb5a6442
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -21,6 +21,7 @@ namespace Emby.Server.Implementations.Session
private readonly SessionInfo _session; private readonly SessionInfo _session;
private readonly List<IWebSocketConnection> _sockets; private readonly List<IWebSocketConnection> _sockets;
private readonly ReaderWriterLockSlim _socketsLock;
private bool _disposed = false; private bool _disposed = false;
public WebSocketController( public WebSocketController(
@ -31,10 +32,26 @@ namespace Emby.Server.Implementations.Session
_logger = logger; _logger = logger;
_session = session; _session = session;
_sessionManager = sessionManager; _sessionManager = sessionManager;
_sockets = new List<IWebSocketConnection>(); _sockets = new();
_socketsLock = new();
} }
private bool HasOpenSockets => GetActiveSockets().Any(); private bool HasOpenSockets
{
get
{
ObjectDisposedException.ThrowIf(_disposed, this);
try
{
_socketsLock.EnterReadLock();
return _sockets.Any(i => i.State == WebSocketState.Open);
}
finally
{
_socketsLock.ExitReadLock();
}
}
}
/// <inheritdoc /> /// <inheritdoc />
public bool SupportsMediaControl => HasOpenSockets; public bool SupportsMediaControl => HasOpenSockets;
@ -42,23 +59,38 @@ namespace Emby.Server.Implementations.Session
/// <inheritdoc /> /// <inheritdoc />
public bool IsSessionActive => HasOpenSockets; public bool IsSessionActive => HasOpenSockets;
private IEnumerable<IWebSocketConnection> GetActiveSockets()
=> _sockets.Where(i => i.State == WebSocketState.Open);
public void AddWebSocket(IWebSocketConnection connection) public void AddWebSocket(IWebSocketConnection connection)
{ {
_logger.LogDebug("Adding websocket to session {Session}", _session.Id); _logger.LogDebug("Adding websocket to session {Session}", _session.Id);
ObjectDisposedException.ThrowIf(_disposed, this);
try
{
_socketsLock.EnterWriteLock();
_sockets.Add(connection); _sockets.Add(connection);
connection.Closed += OnConnectionClosed; connection.Closed += OnConnectionClosed;
} }
finally
{
_socketsLock.ExitWriteLock();
}
}
private async void OnConnectionClosed(object? sender, EventArgs e) private async void OnConnectionClosed(object? sender, EventArgs e)
{ {
var connection = sender as IWebSocketConnection ?? throw new ArgumentException($"{nameof(sender)} is not of type {nameof(IWebSocketConnection)}", nameof(sender)); var connection = sender as IWebSocketConnection ?? throw new ArgumentException($"{nameof(sender)} is not of type {nameof(IWebSocketConnection)}", nameof(sender));
_logger.LogDebug("Removing websocket from session {Session}", _session.Id); _logger.LogDebug("Removing websocket from session {Session}", _session.Id);
ObjectDisposedException.ThrowIf(_disposed, this);
try
{
_socketsLock.EnterWriteLock();
_sockets.Remove(connection); _sockets.Remove(connection);
connection.Closed -= OnConnectionClosed; connection.Closed -= OnConnectionClosed;
}
finally
{
_socketsLock.ExitWriteLock();
}
await _sessionManager.CloseIfNeededAsync(_session).ConfigureAwait(false); await _sessionManager.CloseIfNeededAsync(_session).ConfigureAwait(false);
} }
@ -69,7 +101,17 @@ namespace Emby.Server.Implementations.Session
T data, T data,
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
var socket = GetActiveSockets().MaxBy(i => i.LastActivityDate); ObjectDisposedException.ThrowIf(_disposed, this);
IWebSocketConnection? socket;
try
{
_socketsLock.EnterReadLock();
socket = _sockets.Where(i => i.State == WebSocketState.Open).MaxBy(i => i.LastActivityDate);
}
finally
{
_socketsLock.ExitReadLock();
}
if (socket is null) if (socket is null)
{ {
@ -94,12 +136,23 @@ namespace Emby.Server.Implementations.Session
return; return;
} }
try
{
_socketsLock.EnterWriteLock();
foreach (var socket in _sockets) foreach (var socket in _sockets)
{ {
socket.Closed -= OnConnectionClosed; socket.Closed -= OnConnectionClosed;
socket.Dispose(); socket.Dispose();
} }
_sockets.Clear();
}
finally
{
_socketsLock.ExitWriteLock();
}
_socketsLock.Dispose();
_disposed = true; _disposed = true;
} }
@ -110,12 +163,23 @@ namespace Emby.Server.Implementations.Session
return; return;
} }
try
{
_socketsLock.EnterWriteLock();
foreach (var socket in _sockets) foreach (var socket in _sockets)
{ {
socket.Closed -= OnConnectionClosed; socket.Closed -= OnConnectionClosed;
await socket.DisposeAsync().ConfigureAwait(false); await socket.DisposeAsync().ConfigureAwait(false);
} }
_sockets.Clear();
}
finally
{
_socketsLock.ExitWriteLock();
}
_socketsLock.Dispose();
_disposed = true; _disposed = true;
} }
} }

Loading…
Cancel
Save