// // Copyright (c) PlaceholderCompany. All rights reserved. // namespace Jellyfin.Database.Implementations.Locking; #pragma warning disable MT1013 // Releasing lock without guarantee of execution #pragma warning disable MT1012 // Acquiring lock without guarantee of releasing #pragma warning disable CA1873 #pragma warning disable IDISP004 #pragma warning disable CA1848 #pragma warning disable SA1309 // Variables should not begin with underscore using System; using System.Data; using System.Data.Common; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; /// /// A locking behavior that will always block any operation while a write is requested. Mimicks the old SqliteRepository behavior. /// public class PessimisticLockBehavior : IEntityFrameworkCoreLockingBehavior { private readonly ILogger _logger; private readonly ILoggerFactory _loggerFactory; /// /// Initializes a new instance of the class. /// /// The application logger. /// The logger factory. public PessimisticLockBehavior(ILogger logger, ILoggerFactory loggerFactory) { this._logger = logger; this._loggerFactory = loggerFactory; } private static ReaderWriterLockSlim DatabaseLock { get; } = new(LockRecursionPolicy.SupportsRecursion); private static AsyncLocal WriteLockDepth { get; } = new(); /// public void OnSaveChanges(JellyfinDbContext context, Action saveChanges) { using (DbLock.EnterWrite(this._logger)) { ArgumentNullException.ThrowIfNull(saveChanges); saveChanges(); } } /// public void Initialise(DbContextOptionsBuilder optionsBuilder) { this._logger.LogInformation("The database locking mode has been set to: Pessimistic."); ArgumentNullException.ThrowIfNull(optionsBuilder); optionsBuilder.AddInterceptors(new CommandLockingInterceptor(this._loggerFactory.CreateLogger())); optionsBuilder.AddInterceptors(new TransactionLockingInterceptor(this._loggerFactory.CreateLogger())); } /// public async Task OnSaveChangesAsync(JellyfinDbContext context, Func saveChanges) { using (DbLock.EnterWrite(this._logger)) { ArgumentNullException.ThrowIfNull(saveChanges); await saveChanges().ConfigureAwait(false); } } private sealed class TransactionLockingInterceptor : DbTransactionInterceptor { private readonly ILogger _logger; public TransactionLockingInterceptor(ILogger logger) { this._logger = logger; } public override InterceptionResult TransactionStarting(DbConnection connection, TransactionStartingEventData eventData, InterceptionResult result) { var currentDepth = WriteLockDepth.Value; if (currentDepth == 0) { DbLock.BeginWriteLock(this._logger); WriteLockDepth.Value = currentDepth + 1; } return base.TransactionStarting(connection, eventData, result); } public override ValueTask> TransactionStartingAsync(DbConnection connection, TransactionStartingEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { var currentDepth = WriteLockDepth.Value; if (currentDepth == 0) { DbLock.BeginWriteLock(this._logger); WriteLockDepth.Value = currentDepth + 1; } return base.TransactionStartingAsync(connection, eventData, result, cancellationToken); } public override void TransactionCommitted(DbTransaction transaction, TransactionEndEventData eventData) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } base.TransactionCommitted(transaction, eventData); } public override Task TransactionCommittedAsync(DbTransaction transaction, TransactionEndEventData eventData, CancellationToken cancellationToken = default) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } return base.TransactionCommittedAsync(transaction, eventData, cancellationToken); } public override void TransactionFailed(DbTransaction transaction, TransactionErrorEventData eventData) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } base.TransactionFailed(transaction, eventData); } public override Task TransactionFailedAsync(DbTransaction transaction, TransactionErrorEventData eventData, CancellationToken cancellationToken = default) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } return base.TransactionFailedAsync(transaction, eventData, cancellationToken); } public override void TransactionRolledBack(DbTransaction transaction, TransactionEndEventData eventData) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } base.TransactionRolledBack(transaction, eventData); } public override Task TransactionRolledBackAsync(DbTransaction transaction, TransactionEndEventData eventData, CancellationToken cancellationToken = default) { var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { WriteLockDepth.Value = currentDepth - 1; if (currentDepth == 1) { DbLock.EndWriteLock(this._logger); } } return base.TransactionRolledBackAsync(transaction, eventData, cancellationToken); } } /// /// Adds strict read/write locking. /// private sealed class CommandLockingInterceptor : DbCommandInterceptor { private readonly ILogger _logger; public CommandLockingInterceptor(ILogger logger) { this._logger = logger; } public override InterceptionResult NonQueryExecuting(DbCommand command, CommandEventData eventData, InterceptionResult result) { using (DbLock.EnterWrite(this._logger, command)) { return InterceptionResult.SuppressWithResult(command.ExecuteNonQuery()); } } public override async ValueTask> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { using (DbLock.EnterWrite(this._logger, command)) { return InterceptionResult.SuppressWithResult(await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false)); } } public override InterceptionResult ScalarExecuting(DbCommand command, CommandEventData eventData, InterceptionResult result) { using (DbLock.EnterRead(this._logger)) { return InterceptionResult.SuppressWithResult(command.ExecuteScalar()!); } } public override async ValueTask> ScalarExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { using (DbLock.EnterRead(this._logger)) { return InterceptionResult.SuppressWithResult((await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!); } } public override InterceptionResult ReaderExecuting(DbCommand command, CommandEventData eventData, InterceptionResult result) { using (DbLock.EnterRead(this._logger)) { return InterceptionResult.SuppressWithResult(command.ExecuteReader()!); } } public override async ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) { using (DbLock.EnterRead(this._logger)) { return InterceptionResult.SuppressWithResult(await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false)); } } } private sealed class DbLock : IDisposable { private static readonly IDisposable _noLock = new DbLock(null) { _disposed = true }; private static (string Command, Guid Id, DateTimeOffset QueryDate, bool Printed) _blockQuery; private readonly Action? _action; private bool _disposed; private bool _lockAcquired; public DbLock(Action? action = null) { this._action = action; } #pragma warning disable IDISP015 // Member should not return created and cached instance public static IDisposable EnterWrite(ILogger logger, IDbCommand? command = null, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) #pragma warning restore IDISP015 // Member should not return created and cached instance { logger.LogTrace("Enter Write for {Caller}:{Line}", callerMemberName, callerNo); var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { logger.LogTrace("Write Already Held (Depth: {Depth}) {Caller}:{Line}", currentDepth, callerMemberName, callerNo); return _noLock; } BeginWriteLock(logger, command, callerMemberName, callerNo); WriteLockDepth.Value = currentDepth + 1; return new DbLock(() => { WriteLockDepth.Value = Math.Max(0, WriteLockDepth.Value - 1); EndWriteLock(logger, callerMemberName, callerNo); }) { _lockAcquired = true }; } #pragma warning disable IDISP015 // Member should not return created and cached instance public static IDisposable EnterRead(ILogger logger, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) #pragma warning restore IDISP015 // Member should not return created and cached instance { logger.LogTrace("Enter Read {Caller}:{Line}", callerMemberName, callerNo); var currentDepth = WriteLockDepth.Value; if (currentDepth > 0) { logger.LogTrace("Write Already Held (Depth: {Depth}) {Caller}:{Line}", currentDepth, callerMemberName, callerNo); return _noLock; } BeginReadLock(logger, callerMemberName, callerNo); return new DbLock(() => { ExitReadLock(logger, callerMemberName, callerNo); }) { _lockAcquired = true }; } public static void BeginWriteLock(ILogger logger, IDbCommand? command = null, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) { logger.LogTrace("Aquire Write {Caller}:{Line}", callerMemberName, callerNo); if (!DatabaseLock.TryEnterWriteLock(TimeSpan.FromMilliseconds(1000))) { var blockingQuery = _blockQuery; if (!blockingQuery.Printed) { _blockQuery = (blockingQuery.Command, blockingQuery.Id, blockingQuery.QueryDate, true); logger.LogInformation("QueryLock: {Id} --- {Query}", blockingQuery.Id, blockingQuery.Command); } logger.LogInformation("Query congestion detected: '{Id}' since '{Date}'", blockingQuery.Id, blockingQuery.QueryDate); DatabaseLock.EnterWriteLock(); logger.LogInformation("Query congestion cleared: '{Id}' for '{Date}'", blockingQuery.Id, DateTimeOffset.Now - blockingQuery.QueryDate); } _blockQuery = (command?.CommandText ?? "Transaction", Guid.NewGuid(), DateTimeOffset.Now, false); logger.LogTrace("Write Aquired {Caller}:{Line}", callerMemberName, callerNo); } public static void BeginReadLock(ILogger logger, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) { logger.LogTrace("Aquire Write {Caller}:{Line}", callerMemberName, callerNo); DatabaseLock.EnterReadLock(); logger.LogTrace("Read Aquired {Caller}:{Line}", callerMemberName, callerNo); } public static void EndWriteLock(ILogger logger, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) { logger.LogTrace("Release Write {Caller}:{Line}", callerMemberName, callerNo); DatabaseLock.ExitWriteLock(); } public static void ExitReadLock(ILogger logger, [CallerMemberName] string? callerMemberName = null, [CallerLineNumber] int? callerNo = null) { logger.LogTrace("Release Read {Caller}:{Line}", callerMemberName, callerNo); DatabaseLock.ExitReadLock(); } public void Dispose() { if (this._disposed) { return; } this._disposed = true; if (this._lockAcquired && this._action is not null) { this._action(); } } } }