mirror of
https://github.com/2dust/v2rayN.git
synced 2026-10-18 03:05:00 +03:00
Fix data race in ProfileExManager (#10388)
* Fix data race in ProfileExManager Parallel speed tests over many profiles corrupt the save queue of test results: IndexIdEnqueue touches a plain Queue<string> from every test thread, so Contains plus Enqueue race - the same profile gets queued twice, and Queue.Contains can even throw ArgumentOutOfRangeException out of SetTestDelay, losing the result. SaveQueueIndexIds has the same class of bug: it snapshots Count and then dequeues in a loop, so concurrent enqueues throw or lose entries. GetProfileExItem could create duplicate rows for the same index the same way. Guard the queue and the profile lookup with a lock and drain a snapshot instead of a stale count. No behavior change besides correctness. * Fix data race in ProfileExManager Parallel speed tests over many profiles corrupt the save queue of test results: IndexIdEnqueue touches a plain Queue<string> from every test thread, so Contains plus Enqueue race - the same profile gets queued twice, and Queue.Contains can even throw ArgumentOutOfRangeException out of SetTestDelay, losing the result. SaveQueueIndexIds has the same class of bug: it snapshots Count and then dequeues in a loop, so concurrent enqueues throw or lose entries. Switch the queue to ConcurrentQueue and drain it with TryDequeue. As suggested in review, duplicates are not filtered anymore: this is a helper, and the save loop just writes the same row more than once.
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
namespace ServiceLib.Tests.Manager;
|
||||
|
||||
public class ProfileExManagerConcurrencyTests
|
||||
{
|
||||
[Test]
|
||||
public async Task SetTestDelay_FromManyThreads_ShouldNotThrow()
|
||||
{
|
||||
var manager = CreateManager();
|
||||
var exceptions = new List<Exception>();
|
||||
var threads = Environment.ProcessorCount * 2;
|
||||
|
||||
//Hammer the same manager from several streams at once: the queue is mutated while
|
||||
//other threads keep pushing into it
|
||||
var barrier = new Barrier(threads);
|
||||
Parallel.For(0, threads, t =>
|
||||
{
|
||||
barrier.SignalAndWait();
|
||||
for (var i = 0; i < 5000; i++)
|
||||
{
|
||||
try
|
||||
{
|
||||
manager.SetTestDelay($"index{i % 500}", i);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
lock (exceptions)
|
||||
{
|
||||
exceptions.Add(ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
await exceptions.Count.Should().BeEqualTo(0);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task SetTestDelay_FromManyThreads_ShouldNotLoseQueuedIds()
|
||||
{
|
||||
var manager = CreateManager();
|
||||
var ids = 500;
|
||||
|
||||
Parallel.For(0, Environment.ProcessorCount * 2, t =>
|
||||
{
|
||||
for (var i = 0; i < 5000; i++)
|
||||
{
|
||||
manager.SetTestDelay($"id{i % ids}", i);
|
||||
}
|
||||
});
|
||||
|
||||
var drained = Drain(manager);
|
||||
await drained.Count.Should().BeGreaterThanOrEqualTo(ids);
|
||||
await drained.Distinct().Count().Should().BeEqualTo(ids);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task SetTestDelay_ShouldCreateOneEntryPerIndex()
|
||||
{
|
||||
var manager = CreateManager();
|
||||
|
||||
await Parallel.ForEachAsync(Enumerable.Range(0, 500), new ParallelOptions { MaxDegreeOfParallelism = 32 },
|
||||
async (i, ct) =>
|
||||
{
|
||||
manager.SetTestDelay($"same{i % 10}", i);
|
||||
await Task.CompletedTask;
|
||||
});
|
||||
|
||||
var items = await manager.GetProfileExs();
|
||||
await items.Count.Should().BeEqualTo(10);
|
||||
await items.Select(t => t.IndexId).Distinct().Count().Should().BeEqualTo(10);
|
||||
}
|
||||
|
||||
private static ProfileExManager CreateManager()
|
||||
{
|
||||
return (ProfileExManager)Activator.CreateInstance(typeof(ProfileExManager), nonPublic: true)!;
|
||||
}
|
||||
|
||||
private static List<string> Drain(ProfileExManager manager)
|
||||
{
|
||||
var queue = (System.Collections.Concurrent.ConcurrentQueue<string>)typeof(ProfileExManager)
|
||||
.GetField("_queIndexIds", BindingFlags.Instance | BindingFlags.NonPublic)!
|
||||
.GetValue(manager)!;
|
||||
|
||||
var result = new List<string>();
|
||||
while (queue.TryDequeue(out var id))
|
||||
{
|
||||
result.Add(id);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,7 @@ public class ProfileExManager
|
||||
{
|
||||
private static readonly Lazy<ProfileExManager> _instance = new(() => new());
|
||||
private ConcurrentBag<ProfileExItem> _lstProfileEx = [];
|
||||
private readonly Queue<string> _queIndexIds = new();
|
||||
private readonly ConcurrentQueue<string> _queIndexIds = new();
|
||||
public static ProfileExManager Instance => _instance.Value;
|
||||
private static readonly string _tag = "ProfileExHandler";
|
||||
|
||||
@@ -32,24 +32,30 @@ public class ProfileExManager
|
||||
|
||||
private void IndexIdEnqueue(string indexId)
|
||||
{
|
||||
if (indexId.IsNotEmpty() && !_queIndexIds.Contains(indexId))
|
||||
if (indexId.IsNotEmpty())
|
||||
{
|
||||
//Speed tests touch this from many threads; duplicates are harmless here since
|
||||
//the save loop just writes the same row more than once
|
||||
_queIndexIds.Enqueue(indexId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SaveQueueIndexIds()
|
||||
{
|
||||
var cnt = _queIndexIds.Count;
|
||||
if (cnt > 0)
|
||||
var lstIds = new List<string>();
|
||||
while (_queIndexIds.TryDequeue(out var id))
|
||||
{
|
||||
lstIds.Add(id);
|
||||
}
|
||||
|
||||
if (lstIds.Count > 0)
|
||||
{
|
||||
var lstExists = await SQLiteHelper.Instance.TableAsync<ProfileExItem>().ToListAsync();
|
||||
List<ProfileExItem> lstInserts = [];
|
||||
List<ProfileExItem> lstUpdates = [];
|
||||
|
||||
for (var i = 0; i < cnt; i++)
|
||||
foreach (var id in lstIds)
|
||||
{
|
||||
var id = _queIndexIds.Dequeue();
|
||||
var item = lstExists.FirstOrDefault(t => t.IndexId == id);
|
||||
var itemNew = _lstProfileEx?.FirstOrDefault(t => t.IndexId == id);
|
||||
if (itemNew is null)
|
||||
|
||||
Reference in New Issue
Block a user