From 673283da4b8666c65c258a1949882ec41b6b5cad Mon Sep 17 00:00:00 2001 From: Dmitry Date: Sat, 17 Oct 2026 14:37:43 +0300 Subject: [PATCH] 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 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 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. --- .../ProfileExManagerConcurrencyTests.cs | 91 +++++++++++++++++++ v2rayN/ServiceLib/Manager/ProfileExManager.cs | 18 ++-- 2 files changed, 103 insertions(+), 6 deletions(-) create mode 100644 v2rayN/ServiceLib.Tests/Manager/ProfileExManagerConcurrencyTests.cs diff --git a/v2rayN/ServiceLib.Tests/Manager/ProfileExManagerConcurrencyTests.cs b/v2rayN/ServiceLib.Tests/Manager/ProfileExManagerConcurrencyTests.cs new file mode 100644 index 00000000..a5d97f09 --- /dev/null +++ b/v2rayN/ServiceLib.Tests/Manager/ProfileExManagerConcurrencyTests.cs @@ -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(); + 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 Drain(ProfileExManager manager) + { + var queue = (System.Collections.Concurrent.ConcurrentQueue)typeof(ProfileExManager) + .GetField("_queIndexIds", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(manager)!; + + var result = new List(); + while (queue.TryDequeue(out var id)) + { + result.Add(id); + } + return result; + } +} \ No newline at end of file diff --git a/v2rayN/ServiceLib/Manager/ProfileExManager.cs b/v2rayN/ServiceLib/Manager/ProfileExManager.cs index 44baf951..89d81ac7 100644 --- a/v2rayN/ServiceLib/Manager/ProfileExManager.cs +++ b/v2rayN/ServiceLib/Manager/ProfileExManager.cs @@ -4,7 +4,7 @@ public class ProfileExManager { private static readonly Lazy _instance = new(() => new()); private ConcurrentBag _lstProfileEx = []; - private readonly Queue _queIndexIds = new(); + private readonly ConcurrentQueue _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(); + while (_queIndexIds.TryDequeue(out var id)) + { + lstIds.Add(id); + } + + if (lstIds.Count > 0) { var lstExists = await SQLiteHelper.Instance.TableAsync().ToListAsync(); List lstInserts = []; List 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)