From 337bd03c14f098a4d8f8fee521fa71842092c5eb Mon Sep 17 00:00:00 2001 From: Frank Date: Fri, 16 Oct 2026 21:43:30 +0000 Subject: [PATCH] Optimize in-memory job acquisition and queries --- .../Storage/InMemoryJobStorage.Acquire.cs | 117 +++++++++------ .../Storage/InMemoryJobStorage.Query.cs | 15 +- ...MemoryStoragePerformanceRegressionTests.cs | 140 ++++++++++++++++++ 3 files changed, 222 insertions(+), 50 deletions(-) create mode 100644 tests/Immediate.Jobs.StorageTests/InMemoryStoragePerformanceRegressionTests.cs diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs index 8d935c31..dc9b4e5d 100644 --- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs +++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs @@ -119,6 +119,7 @@ private void AcquireQueueInExistingOrder( List acquired ) { + var eligibility = new Dictionary(StringComparer.Ordinal); foreach (var candidate in _jobs.Values .Where(job => string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) && jobCapacities.ContainsKey(job.JobName) && @@ -130,10 +131,11 @@ job.State is JobState.Pending or JobState.Scheduled && { if (queueCapacity == 0) break; - if (jobCapacities[candidate.JobName] <= 0 || !CanAcquire(candidate.JobName, now)) + if (jobCapacities[candidate.JobName] <= 0 || !CanAcquire(candidate.JobName, now, eligibility)) continue; acquired.Add(Acquire(candidate, request, now)); + eligibility.Remove(candidate.JobName); jobCapacities[candidate.JobName]--; queueCapacity--; } @@ -161,62 +163,89 @@ List acquired ) { var policy = request.FairQueues!; + // The gate and acquisition timestamp stay fixed for the entire batch. Build ordered + // backlogs once, then advance past definitions whose capacity or limits are exhausted. + var groups = _jobs.Values + .Where(job => string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) && + jobCapacities.TryGetValue(job.JobName, out var capacity) && capacity > 0 && + job.State is JobState.Pending or JobState.Scheduled && job.DueAt <= now) + .OrderBy(static job => job.DueAt) + .ThenBy(static job => job.CreatedAt) + .ThenBy(static job => job.JobHandle.Value, StringComparer.Ordinal) + .GroupBy(static job => job.GroupId, StringComparer.Ordinal) + .Select(static group => new Queue(group)) + .ToList(); + var eligibility = new Dictionary(StringComparer.Ordinal); + var activeCounts = new Dictionary(StringComparer.Ordinal); + var totalActive = 0; + foreach (var job in _jobs.Values) + { + if (!string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) || + job.State != JobState.Active || job.LeaseExpiresAt is null || job.LeaseExpiresAt <= now) + { + continue; + } + + totalActive++; + if (job.GroupId is { } groupId) + activeCounts[groupId] = activeCounts.GetValueOrDefault(groupId) + 1; + } + + var nextSequence = policy.GroupRoundRobin ? GetNextSequence(queueName) : 0; + var candidates = new List>(groups.Count); while (queueCapacity > 0) { - var eligible = _jobs.Values - .Where(job => string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) && - jobCapacities.TryGetValue(job.JobName, out var capacity) && - capacity > 0 && - job.State is JobState.Pending or JobState.Scheduled && - job.DueAt <= now && CanAcquire(job.JobName, now)) - .ToList(); - if (eligible.Count == 0) + candidates.Clear(); + foreach (var group in groups) + { + while (group.TryPeek(out var head)) + { + if (jobCapacities[head.JobName] > 0 && CanAcquire(head.JobName, now, eligibility)) + { + candidates.Add(group); + break; + } + + group.Dequeue(); + } + } + + if (candidates.Count == 0) break; - var activeCounts = _jobs.Values - .Where(job => string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) && - job.GroupId is not null && - job.State == JobState.Active && - job.LeaseExpiresAt > now) - .GroupBy(static job => job.GroupId!, StringComparer.Ordinal) - .ToDictionary(static group => group.Key, static group => group.Count(), StringComparer.Ordinal); - var totalActive = _jobs.Values.Count(job => string.Equals(GetQueueName(job.JobName), queueName, StringComparison.Ordinal) && - job.State == JobState.Active && - job.LeaseExpiresAt > now); - var groupedHeads = eligible - .Where(static job => job.GroupId is not null) - .GroupBy(static job => job.GroupId!, StringComparer.Ordinal) - .Select(static group => group - .OrderBy(static job => job.DueAt) - .ThenBy(static job => job.CreatedAt) - .ThenBy(static job => job.JobHandle.Value, StringComparer.Ordinal) - .First()); - var ungroupedHead = eligible - .Where(static job => job.GroupId is null) - .OrderBy(static job => job.DueAt) - .ThenBy(static job => job.CreatedAt) - .ThenBy(static job => job.JobHandle.Value, StringComparer.Ordinal) - .Take(1); - var candidates = groupedHeads.Concat(ungroupedHead); - - var candidate = candidates - .OrderBy(job => IsNoisy(job.GroupId, activeCounts, totalActive, policy)) - .ThenBy(job => GetNoisyInflight(job.GroupId, activeCounts, totalActive, policy)) - .ThenBy(job => policy.GroupRoundRobin ? GetLastServed(queueName, job.GroupId) : 0) - .ThenBy(static job => job.DueAt) - .ThenBy(static job => job.CreatedAt) - .ThenBy(static job => job.JobHandle.Value, StringComparer.Ordinal) + var selected = candidates + .OrderBy(group => IsNoisy(group.Peek().GroupId, activeCounts, totalActive, policy)) + .ThenBy(group => GetNoisyInflight(group.Peek().GroupId, activeCounts, totalActive, policy)) + .ThenBy(group => policy.GroupRoundRobin ? GetLastServed(queueName, group.Peek().GroupId) : 0) + .ThenBy(static group => group.Peek().DueAt) + .ThenBy(static group => group.Peek().CreatedAt) + .ThenBy(static group => group.Peek().JobHandle.Value, StringComparer.Ordinal) .First(); - var job = Acquire(candidate, request, now); + var job = Acquire(selected.Dequeue(), request, now); acquired.Add(job); + eligibility.Remove(job.JobName); jobCapacities[job.JobName]--; queueCapacity--; + if (job.LeaseExpiresAt > now) + { + totalActive++; + if (job.GroupId is { } activeGroup) + activeCounts[activeGroup] = activeCounts.GetValueOrDefault(activeGroup) + 1; + } + if (policy.GroupRoundRobin && job.GroupId is { } groupId) - _fairQueueLastServed[(queueName, groupId)] = GetNextSequence(queueName); + _fairQueueLastServed[(queueName, groupId)] = nextSequence++; } } + private bool CanAcquire(string jobName, DateTimeOffset now, Dictionary eligibility) + { + if (!eligibility.TryGetValue(jobName, out var ready)) + eligibility.Add(jobName, ready = CanAcquire(jobName, now)); + return ready; + } + private static bool IsNoisy( string? groupId, Dictionary activeCounts, diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs index c4d78b6c..8c3d16ec 100644 --- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs +++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs @@ -50,7 +50,9 @@ public async ValueTask GetMonitoringSnapshotAsync(Cancell lock (_gate) { - var counts = Enum.GetValues().ToDictionary(state => state, state => _jobs.Values.LongCount(x => x.State == state)); + var counts = Enum.GetValues().ToDictionary(static state => state, static _ => 0L); + foreach (var job in _jobs.Values) + counts[job.State]++; var now = timeProvider.GetUtcNow(); var servers = GetLiveServersSnapshot(); return new JobMonitoringSnapshot @@ -74,13 +76,13 @@ public async ValueTask> QueryJobsAsync(JobQuery query, lock (_gate) { - var jobs = _jobs.Values.Select(ReadRecord); + IEnumerable jobs = _jobs.Values; if (query.JobHandle is { } id) - jobs = jobs.Where(x => x.JobHandle == query.JobHandle); + jobs = _jobs.TryGetValue(id, out var job) ? [job] : []; if (query.State is { } state) jobs = jobs.Where(x => x.State == state); if (!string.IsNullOrWhiteSpace(query.QueueName)) - jobs = jobs.Where(x => string.Equals(x.QueueName, query.QueueName, StringComparison.Ordinal)); + jobs = jobs.Where(x => string.Equals(GetQueueName(x.JobName), query.QueueName, StringComparison.Ordinal)); if (!string.IsNullOrWhiteSpace(query.JobName)) jobs = jobs.Where(x => string.Equals(x.JobName, query.JobName, StringComparison.OrdinalIgnoreCase)); if (query.CreatedBefore is { } createdBefore) @@ -94,7 +96,8 @@ public async ValueTask> QueryJobsAsync(JobQuery query, .. jobs.OrderByDescending(x => x.CreatedAt) .ThenBy(x => x.JobHandle.Value, StringComparer.Ordinal) .Skip(Math.Max(0, query.Skip)) - .Take(Math.Clamp(query.Take, 1, 1000)), + .Take(Math.Clamp(query.Take, 1, 1000)) + .Select(ReadRecord), ]; } } @@ -109,7 +112,6 @@ public async ValueTask> QueryNonCompletedJobsAsync(stri lock (_gate) { return _jobs.Values - .Select(ReadRecord) .Where(j => string.Equals(j.JobName, jobName, StringComparison.OrdinalIgnoreCase)) .Where( j => j.State is @@ -119,6 +121,7 @@ or JobState.Scheduled or JobState.Pending or JobState.Active ) + .Select(ReadRecord) .ToList(); } } diff --git a/tests/Immediate.Jobs.StorageTests/InMemoryStoragePerformanceRegressionTests.cs b/tests/Immediate.Jobs.StorageTests/InMemoryStoragePerformanceRegressionTests.cs new file mode 100644 index 00000000..d247fcb3 --- /dev/null +++ b/tests/Immediate.Jobs.StorageTests/InMemoryStoragePerformanceRegressionTests.cs @@ -0,0 +1,140 @@ +using System.Globalization; +using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Storage; +using Microsoft.Extensions.Time.Testing; + +namespace Immediate.Jobs.StorageTests; + +public sealed class InMemoryStoragePerformanceRegressionTests +{ + [Theory] + [InlineData(1, 0, 0, 10)] + [InlineData(0, 1, 0, 10)] + [InlineData(0, 0, 1, 10)] + [InlineData(0, 0, 0, 1)] + public async Task FairAcquisitionAdvancesPastExhaustedDefinitionsWithinEachGroup( + int maxConcurrency, int slidingMax, int fixedMax, int capacity + ) + { + var token = TestContext.Current.CancellationToken; + var clock = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + await using var storage = new InMemoryJobStorage(clock); + await storage.MergeJobDefinitionsListAsync(new() + { + Definitions = + [ + new() + { + Name = "limited", + MaxConcurrency = maxConcurrency, + SlidingWindowMax = slidingMax, + SlidingWindowPeriod = slidingMax > 0 ? TimeSpan.FromMinutes(1) : null, + FixedWindowMax = fixedMax, + FixedWindowPeriod = fixedMax > 0 ? TimeSpan.FromMinutes(1) : null, + }, + new() { Name = "free" }, + ], + }, token); + (string Id, string Name, string? Group)[] jobs = + [ + ("limited-a", "limited", "a"), + ("limited-a-next", "limited", "a"), + ("free-a", "free", "a"), + ("limited-b", "limited", "b"), + ("free-b", "free", "b"), + ("limited-ungrouped", "limited", null), + ("free-ungrouped", "free", null), + ]; + for (var index = 0; index < jobs.Length; index++) + { + var (id, name, group) = jobs[index]; + await storage.EnqueueAsync(Job(id, name, group, clock.GetUtcNow().AddTicks(index)), token); + } + + var acquired = await storage.AcquireDueJobsAsync(Request(capacity, roundRobin: true), token); + + Assert.Equal(["limited-a", "free-b", "free-ungrouped", "free-a"], acquired.Select(static job => job.JobHandle.Value)); + } + + [Fact] + public async Task FairAcquisitionReclassifiesNoisyGroupsAfterEachClaim() + { + var token = TestContext.Current.CancellationToken; + var clock = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + await using var storage = new InMemoryJobStorage(clock); + await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "free" }] }, token); + for (var index = 0; index < 6; index++) + { + var group = index < 3 ? "a" : "b"; + var id = string.Create(CultureInfo.InvariantCulture, $"{group}-{index % 3}"); + await storage.EnqueueAsync(Job(id, "free", group, clock.GetUtcNow().AddTicks(index)), token); + } + + var acquired = await storage.AcquireDueJobsAsync(Request(10, roundRobin: false), token); + + Assert.Equal(["a-0", "b-0", "a-1", "b-1", "a-2", "b-2"], acquired.Select(static job => job.JobHandle.Value)); + } + + [Fact] + public async Task ExactJobQueryComposesFiltersAndUsesCurrentQueue() + { + var token = TestContext.Current.CancellationToken; + var clock = new FakeTimeProvider(DateTimeOffset.UnixEpoch); + await using var storage = new InMemoryJobStorage(clock); + await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "free", QueueName = "old" }] }, token); + await storage.EnqueueAsync(Job("target", "free", group: null, clock.GetUtcNow()), token); + await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "free", QueueName = "current" }] }, token); + var query = new JobQuery + { + JobHandle = JobHandle.FromString("target"), + State = JobState.Pending, + JobName = "FREE", + Search = "REE", + QueueName = "current", + CreatedBefore = clock.GetUtcNow().AddTicks(1), + }; + + Assert.Equal("current", Assert.Single(await storage.QueryJobsAsync(query, token)).QueueName); + Assert.Equal("current", Assert.Single(await storage.QueryNonCompletedJobsAsync("FREE", token)).QueueName); + JobQuery[] excluded = + [ + query with { JobHandle = JobHandle.FromString("missing") }, + query with { State = JobState.Scheduled }, + query with { JobName = "other" }, + query with { Search = "other" }, + query with { QueueName = "old" }, + query with { CreatedBefore = clock.GetUtcNow() }, + query with { Skip = 1 }, + ]; + foreach (var filter in excluded) + Assert.Empty(await storage.QueryJobsAsync(filter, token)); + } + + private static JobSubmission Job(string id, string name, string? group, DateTimeOffset createdAt) => new() + { + JobHandle = JobHandle.FromString(id), + JobName = name, + Payload = "{}", + State = JobState.Pending, + GroupId = group, + CreatedAt = createdAt, + DueAt = DateTimeOffset.UnixEpoch, + }; + + private static JobAcquisitionRequest Request(int limitedCapacity, bool roundRobin) => new() + { + WorkerId = "worker", + Lease = TimeSpan.FromMinutes(1), + BatchSize = 10, + FairQueues = new() { GroupRoundRobin = roundRobin, ConcurrencyShareThreshold = 0.5, MinInflightForNoisy = roundRobin ? 30 : 1 }, + Queues = + [ + new() + { + QueueName = "default", + Capacity = 10, + JobCapacities = new Dictionary(StringComparer.Ordinal) { ["LIMITED"] = limitedCapacity, ["free"] = 10 }, + }, + ], + }; +}