1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
using System;
using System.Data.Common;
using System.Linq;
using System.Security.Cryptography;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using Microsoft.Extensions.Logging;
using Polly;
namespace Jellyfin.Database.Implementations.Locking;
/// <summary>
/// Defines a locking mechanism that will retry any write operation for a few times.
/// </summary>
public class OptimisticLockBehavior : IEntityFrameworkCoreLockingBehavior
{
private readonly Policy _writePolicy;
private readonly AsyncPolicy _writeAsyncPolicy;
private readonly ILogger<OptimisticLockBehavior> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="OptimisticLockBehavior"/> class.
/// </summary>
/// <param name="logger">The application logger.</param>
public OptimisticLockBehavior(ILogger<OptimisticLockBehavior> logger)
{
TimeSpan[] sleepDurations = [
TimeSpan.FromMilliseconds(50),
TimeSpan.FromMilliseconds(50),
TimeSpan.FromMilliseconds(50),
TimeSpan.FromMilliseconds(50),
TimeSpan.FromMilliseconds(250),
TimeSpan.FromMilliseconds(250),
TimeSpan.FromMilliseconds(250),
TimeSpan.FromMilliseconds(150),
TimeSpan.FromMilliseconds(150),
TimeSpan.FromMilliseconds(150),
TimeSpan.FromMilliseconds(500),
TimeSpan.FromMilliseconds(150),
TimeSpan.FromMilliseconds(500),
TimeSpan.FromMilliseconds(150),
TimeSpan.FromSeconds(3)
];
Func<int, Context, TimeSpan> backoffProvider = (index, context) =>
{
var backoff = sleepDurations[index];
return backoff + TimeSpan.FromMilliseconds(RandomNumberGenerator.GetInt32(0, (int)(backoff.TotalMilliseconds * .5)));
};
_logger = logger;
_writePolicy = Policy
.HandleInner<Exception>(e => e.Message.Contains("database is locked", StringComparison.InvariantCultureIgnoreCase))
.WaitAndRetry(sleepDurations.Length, backoffProvider, RetryHandle);
_writeAsyncPolicy = Policy
.HandleInner<Exception>(e => e.Message.Contains("database is locked", StringComparison.InvariantCultureIgnoreCase))
.WaitAndRetryAsync(sleepDurations.Length, backoffProvider, RetryHandle);
void RetryHandle(Exception exception, TimeSpan timespan, int retryNo, Context context)
{
if (retryNo < sleepDurations.Length)
{
_logger.LogWarning("Operation failed retry {RetryNo}", retryNo);
}
else
{
_logger.LogError(exception, "Operation failed retry {RetryNo}", retryNo);
}
}
}
/// <inheritdoc/>
public void Initialise(DbContextOptionsBuilder optionsBuilder)
{
_logger.LogInformation("The database locking mode has been set to: Optimistic.");
optionsBuilder.AddInterceptors(new RetryInterceptor(_writeAsyncPolicy, _writePolicy));
optionsBuilder.AddInterceptors(new TransactionLockingInterceptor(_writeAsyncPolicy, _writePolicy));
}
/// <inheritdoc/>
public void OnSaveChanges(JellyfinDbContext context, Action saveChanges)
{
_writePolicy.ExecuteAndCapture(saveChanges);
}
/// <inheritdoc/>
public async Task OnSaveChangesAsync(JellyfinDbContext context, Func<Task> saveChanges)
{
await _writeAsyncPolicy.ExecuteAndCaptureAsync(saveChanges).ConfigureAwait(false);
}
private sealed class TransactionLockingInterceptor : DbTransactionInterceptor
{
private readonly AsyncPolicy _asyncRetryPolicy;
private readonly Policy _retryPolicy;
public TransactionLockingInterceptor(AsyncPolicy asyncRetryPolicy, Policy retryPolicy)
{
_asyncRetryPolicy = asyncRetryPolicy;
_retryPolicy = retryPolicy;
}
public override InterceptionResult<DbTransaction> TransactionStarting(DbConnection connection, TransactionStartingEventData eventData, InterceptionResult<DbTransaction> result)
{
return InterceptionResult<DbTransaction>.SuppressWithResult(_retryPolicy.Execute(() => connection.BeginTransaction(eventData.IsolationLevel)));
}
public override async ValueTask<InterceptionResult<DbTransaction>> TransactionStartingAsync(DbConnection connection, TransactionStartingEventData eventData, InterceptionResult<DbTransaction> result, CancellationToken cancellationToken = default)
{
return InterceptionResult<DbTransaction>.SuppressWithResult(await _asyncRetryPolicy.ExecuteAsync(async () => await connection.BeginTransactionAsync(eventData.IsolationLevel, cancellationToken).ConfigureAwait(false)).ConfigureAwait(false));
}
}
private sealed class RetryInterceptor : DbCommandInterceptor
{
private readonly AsyncPolicy _asyncRetryPolicy;
private readonly Policy _retryPolicy;
public RetryInterceptor(AsyncPolicy asyncRetryPolicy, Policy retryPolicy)
{
_asyncRetryPolicy = asyncRetryPolicy;
_retryPolicy = retryPolicy;
}
public override InterceptionResult<int> NonQueryExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<int> result)
{
return InterceptionResult<int>.SuppressWithResult(_retryPolicy.Execute(command.ExecuteNonQuery));
}
public override async ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<int> result, CancellationToken cancellationToken = default)
{
return InterceptionResult<int>.SuppressWithResult(await _asyncRetryPolicy.ExecuteAsync(async () => await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false)).ConfigureAwait(false));
}
public override InterceptionResult<object> ScalarExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<object> result)
{
return InterceptionResult<object>.SuppressWithResult(_retryPolicy.Execute(() => command.ExecuteScalar()!));
}
public override async ValueTask<InterceptionResult<object>> ScalarExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<object> result, CancellationToken cancellationToken = default)
{
return InterceptionResult<object>.SuppressWithResult((await _asyncRetryPolicy.ExecuteAsync(async () => await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false)!).ConfigureAwait(false))!);
}
public override InterceptionResult<DbDataReader> ReaderExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
{
return InterceptionResult<DbDataReader>.SuppressWithResult(_retryPolicy.Execute(command.ExecuteReader));
}
public override async ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result, CancellationToken cancellationToken = default)
{
return InterceptionResult<DbDataReader>.SuppressWithResult(await _asyncRetryPolicy.ExecuteAsync(async () => await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false)).ConfigureAwait(false));
}
}
}
|
