Serialize player context mutations against the periodic save
A player's progress is persisted from two unrelated flows: action handlers triggered by incoming packets (sequential per connection) and the periodic save, which runs on an independent timer. Action handlers mutate tracked entities with plain field and collection writes - for example crafting toggles item.ItemOptions, and item stacking and NPC selling delete item rows - which bypass the persistence context's own lock. When such a mutation runs while SaveChangesAsync enumerates the change tracker, the save throws (collection-modified, or a DbUpdateConcurrency "affected 0 rows" since the context has no concurrency tokens). SaveChanges is atomic, so every following save fails too and the whole session never persists: on relog the player rolls back, losing progress and items. Add a per-player re-entrant persistence lock and acquire it around both the packet-handling funnel and SaveProgressAsync, so a player's mutations and saves can never overlap. The lock is re-entrant per asynchronous flow (an instance AsyncLocal), so an inline save inside an already-serialized handler does not deadlock; and it is per player, so a trade still acquires the trading partner's lock separately. Add regression tests covering mutual exclusion and re-entrancy of the lock.
This commit is contained in:
@@ -54,6 +54,21 @@ public class Player : AsyncDisposable, IBucketMapObserver, IAttackable, IAttacke
|
|||||||
private readonly AsyncLock _moveLock = new();
|
private readonly AsyncLock _moveLock = new();
|
||||||
private readonly AsyncLock _experienceLock = new();
|
private readonly AsyncLock _experienceLock = new();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Serializes context mutations done by this player's action handlers against the periodic and
|
||||||
|
/// disconnect progress saves, which run on an independent timer flow. See
|
||||||
|
/// <see cref="RunPersistenceExclusiveAsync{T}"/>.
|
||||||
|
/// </summary>
|
||||||
|
private readonly AsyncLock _persistenceLock = new();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Tracks, per asynchronous flow, whether <see cref="_persistenceLock"/> is already held, so the
|
||||||
|
/// lock can be re-entered (Nito's <see cref="AsyncLock"/> is not reentrant). It is an instance
|
||||||
|
/// field on purpose: reentrancy must be tracked per player, so a flow holding player A's lock
|
||||||
|
/// still acquires player B's lock (e.g. during a trade) instead of wrongly skipping it.
|
||||||
|
/// </summary>
|
||||||
|
private readonly AsyncLocal<bool> _persistenceLockHeld = new();
|
||||||
|
|
||||||
private readonly Walker _walker;
|
private readonly Walker _walker;
|
||||||
|
|
||||||
private readonly AppearanceDataAdapter _appearanceData;
|
private readonly AppearanceDataAdapter _appearanceData;
|
||||||
@@ -1879,12 +1894,78 @@ public class Player : AsyncDisposable, IBucketMapObserver, IAttackable, IAttacke
|
|||||||
/// <returns>Success of the save operation.</returns>
|
/// <returns>Success of the save operation.</returns>
|
||||||
public async ValueTask<bool> SaveProgressAsync(CancellationToken cancellationToken = default)
|
public async ValueTask<bool> SaveProgressAsync(CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
if (!this.IsTemplatePlayer)
|
if (this.IsTemplatePlayer)
|
||||||
{
|
{
|
||||||
return await this.PersistenceContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
return true;
|
return await this.RunPersistenceExclusiveAsync(
|
||||||
|
() => this.PersistenceContext.SaveChangesAsync(cancellationToken),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Runs the given operation while holding this player's persistence lock, so that context
|
||||||
|
/// mutations and progress saves for the player never run concurrently.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The periodic progress save (<see cref="PlugIns.PeriodicSaveProgressPlugIn"/>) runs on an
|
||||||
|
/// independent timer flow. Action handlers mutate tracked entities with plain field/collection
|
||||||
|
/// writes (e.g. crafting toggling <c>item.ItemOptions</c>) which bypass the persistence context's
|
||||||
|
/// own lock; if such a mutation runs while <see cref="IContext.SaveChangesAsync"/> enumerates the
|
||||||
|
/// change tracker, the save throws (collection-modified / DbUpdateConcurrency) and every following
|
||||||
|
/// save fails too, so the whole session is lost on relog. Serializing the packet handler funnel
|
||||||
|
/// and the save against each other closes that window. The lock is re-entrant per asynchronous
|
||||||
|
/// flow, so an inline save inside an already-serialized handler does not deadlock.
|
||||||
|
/// </remarks>
|
||||||
|
/// <typeparam name="T">The result type of the operation.</typeparam>
|
||||||
|
/// <param name="operation">The operation to run exclusively.</param>
|
||||||
|
/// <param name="cancellationToken">The cancellation token.</param>
|
||||||
|
/// <returns>The result of the operation.</returns>
|
||||||
|
public async ValueTask<T> RunPersistenceExclusiveAsync<T>(Func<ValueTask<T>> operation, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
if (this._persistenceLockHeld.Value)
|
||||||
|
{
|
||||||
|
return await operation().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
using var l = await this._persistenceLock.LockAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
this._persistenceLockHeld.Value = true;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return await operation().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
this._persistenceLockHeld.Value = false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Runs the given operation while holding this player's persistence lock.
|
||||||
|
/// See <see cref="RunPersistenceExclusiveAsync{T}"/> for the rationale.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="operation">The operation to run exclusively.</param>
|
||||||
|
/// <param name="cancellationToken">The cancellation token.</param>
|
||||||
|
/// <returns>A value task which completes when the operation completed.</returns>
|
||||||
|
public async ValueTask RunPersistenceExclusiveAsync(Func<ValueTask> operation, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
if (this._persistenceLockHeld.Value)
|
||||||
|
{
|
||||||
|
await operation().ConfigureAwait(false);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
using var l = await this._persistenceLock.LockAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
this._persistenceLockHeld.Value = true;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await operation().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
this._persistenceLockHeld.Value = false;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -151,7 +151,7 @@ public class RemotePlayer : Player, IClientVersionProvider, IHasIpAddress
|
|||||||
this.Logger.LogDebug("[C->S] {0}", buffer.ToArray().AsString());
|
this.Logger.LogDebug("[C->S] {0}", buffer.ToArray().AsString());
|
||||||
}
|
}
|
||||||
|
|
||||||
await this.MainPacketHandler.HandlePacketAsync(this, buffer).ConfigureAwait(false);
|
await this.RunPersistenceExclusiveAsync(() => this.MainPacketHandler.HandlePacketAsync(this, buffer)).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
finally
|
finally
|
||||||
{
|
{
|
||||||
|
|||||||
114
tests/MUnique.OpenMU.Tests/PersistenceLockTest.cs
Normal file
114
tests/MUnique.OpenMU.Tests/PersistenceLockTest.cs
Normal file
@@ -0,0 +1,114 @@
|
|||||||
|
// <copyright file="PersistenceLockTest.cs" company="MUnique">
|
||||||
|
// Licensed under the MIT License. See LICENSE file in the project root for full license information.
|
||||||
|
// </copyright>
|
||||||
|
|
||||||
|
namespace MUnique.OpenMU.Tests;
|
||||||
|
|
||||||
|
using System.Threading;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Tests for the per-player persistence lock (<see cref="MUnique.OpenMU.GameLogic.Player.RunPersistenceExclusiveAsync(System.Func{System.Threading.Tasks.ValueTask},System.Threading.CancellationToken)"/>),
|
||||||
|
/// which serializes a player's context mutations against its periodic/disconnect progress saves so they
|
||||||
|
/// can never run concurrently. Without it, a mutation running during <c>SaveChangesAsync</c> corrupts the
|
||||||
|
/// change tracker and rolls the whole session back.
|
||||||
|
/// </summary>
|
||||||
|
[TestFixture]
|
||||||
|
public class PersistenceLockTest
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Verifies that concurrent exclusive operations for the same player never overlap.
|
||||||
|
/// </summary>
|
||||||
|
[Test]
|
||||||
|
public async Task ConcurrentAccessIsSerializedAsync()
|
||||||
|
{
|
||||||
|
var player = await PlayerTestHelper.CreatePlayerAsync().ConfigureAwait(false);
|
||||||
|
var concurrent = 0;
|
||||||
|
var overlapDetected = false;
|
||||||
|
|
||||||
|
async ValueTask BodyAsync()
|
||||||
|
{
|
||||||
|
if (Interlocked.Increment(ref concurrent) > 1)
|
||||||
|
{
|
||||||
|
overlapDetected = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
await Task.Delay(1).ConfigureAwait(false);
|
||||||
|
Interlocked.Decrement(ref concurrent);
|
||||||
|
}
|
||||||
|
|
||||||
|
var tasks = Enumerable.Range(0, 50)
|
||||||
|
.Select(_ => player.RunPersistenceExclusiveAsync(BodyAsync).AsTask())
|
||||||
|
.ToArray();
|
||||||
|
await Task.WhenAll(tasks).ConfigureAwait(false);
|
||||||
|
|
||||||
|
Assert.That(overlapDetected, Is.False, "Two exclusive operations for the same player ran at the same time.");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Verifies that re-entering the lock from within an already-held exclusive scope does not deadlock
|
||||||
|
/// (an inline save inside a packet handler is exactly this case).
|
||||||
|
/// </summary>
|
||||||
|
[Test]
|
||||||
|
public async Task ReentrantAccessDoesNotDeadlockAsync()
|
||||||
|
{
|
||||||
|
var player = await PlayerTestHelper.CreatePlayerAsync().ConfigureAwait(false);
|
||||||
|
var executed = 0;
|
||||||
|
|
||||||
|
var run = player.RunPersistenceExclusiveAsync(async () =>
|
||||||
|
{
|
||||||
|
Interlocked.Increment(ref executed);
|
||||||
|
await player.RunPersistenceExclusiveAsync(async () =>
|
||||||
|
{
|
||||||
|
Interlocked.Increment(ref executed);
|
||||||
|
await Task.Yield();
|
||||||
|
}).ConfigureAwait(false);
|
||||||
|
}).AsTask();
|
||||||
|
|
||||||
|
// If reentrancy deadlocked, this would hang; fail fast instead of blocking the suite.
|
||||||
|
await run.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||||||
|
|
||||||
|
Assert.That(executed, Is.EqualTo(2));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Verifies that a re-entrant exclusive operation still runs while another flow holds the lock:
|
||||||
|
/// the outer flow keeps the lock, an independent flow must wait, and the re-entrant call inside the
|
||||||
|
/// outer flow proceeds without waiting for itself.
|
||||||
|
/// </summary>
|
||||||
|
[Test]
|
||||||
|
public async Task IndependentFlowWaitsWhileLockIsHeldAsync()
|
||||||
|
{
|
||||||
|
var player = await PlayerTestHelper.CreatePlayerAsync().ConfigureAwait(false);
|
||||||
|
var otherEntered = false;
|
||||||
|
var holderHasLock = new TaskCompletionSource();
|
||||||
|
var mayRelease = new TaskCompletionSource();
|
||||||
|
|
||||||
|
// Holder runs on its own flow and keeps the lock until signalled.
|
||||||
|
var holder = Task.Run(() => player.RunPersistenceExclusiveAsync(async () =>
|
||||||
|
{
|
||||||
|
holderHasLock.SetResult();
|
||||||
|
|
||||||
|
// A re-entrant call from the holding flow must NOT block on the lock we already hold.
|
||||||
|
await player.RunPersistenceExclusiveAsync(() => ValueTask.CompletedTask).ConfigureAwait(false);
|
||||||
|
|
||||||
|
await mayRelease.Task.ConfigureAwait(false);
|
||||||
|
}).AsTask());
|
||||||
|
|
||||||
|
await holderHasLock.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||||||
|
|
||||||
|
// Competing flow started from an INDEPENDENT context (does not inherit the reentrancy flag).
|
||||||
|
var other = Task.Run(() => player.RunPersistenceExclusiveAsync(() =>
|
||||||
|
{
|
||||||
|
otherEntered = true;
|
||||||
|
return ValueTask.CompletedTask;
|
||||||
|
}).AsTask());
|
||||||
|
|
||||||
|
await Task.Delay(50).ConfigureAwait(false);
|
||||||
|
Assert.That(otherEntered, Is.False, "An independent flow entered while the lock was held.");
|
||||||
|
|
||||||
|
mayRelease.SetResult();
|
||||||
|
await holder.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||||||
|
await other.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||||||
|
Assert.That(otherEntered, Is.True, "The competing flow never ran after the lock was released.");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user