address quequing slowdown

Author: BMurri

src/TesApi.Web/BatchScheduler.cs

@@ -64,7 +64,7 @@ public partial class BatchScheduler : IBatchScheduler
 
         internal const string NodeTaskRunnerFilename = "tes-runner";
 
-        internal static TimeSpan QueuedTesTaskTaskGroupGatherWindow = TimeSpan.FromSeconds(10);
+        internal static TimeSpan QueuedTesTaskTaskGroupGatherWindow = TimeSpan.FromSeconds(2);
         internal static TimeSpan QueuedTesTaskPoolGroupGatherWindow = TaskScheduler.BatchRunInterval;
 
         private const string AzureSupportUrl = "https://portal.azure.com/#blade/Microsoft_Azure_Support/HelpAndSupportBlade/newsupportrequest";
@@ -233,7 +233,7 @@ async Task<bool> SetTaskStateAndLogAsync(TesTask tesTask, TesState newTaskState,
 
                         if (!pool?.OrphanedTesTasks.TryRemove(tesTask.Id, out _) ?? false)
                         {
-                            pool.AssociatedTesTasks.Where(pair => tesTask.Id.Equals(pair.Value, StringComparison.InvariantCultureIgnoreCase)).ForEach(pair => _ = pool.AssociatedTesTasks.TryRemove(pair));
+                            pool.AssociatedTesTasks.Where(pair => tesTask.Id.Equals(pair.Value, StringComparison.Ordinal)).ForEach(pair => _ = pool.AssociatedTesTasks.TryRemove(pair));
                         }
                     }
 

src/TesApi.Web/TaskScheduler.cs

@@ -56,11 +56,11 @@ internal class TaskScheduler(RunnerEventsProcessor nodeEventProcessor, Microsoft
         , ITaskScheduler
     {
         private static readonly TimeSpan blobRunInterval = TimeSpan.FromSeconds(15);
-        private static readonly TimeSpan queuedRunInterval = TimeSpan.FromMilliseconds(100);
+        private static readonly TimeSpan queuedRunInterval = TimeSpan.FromSeconds(15);
         private static readonly TimeSpan queuedRepositoryInterval = TimeSpan.FromMinutes(1);
         internal static readonly TimeSpan BatchRunInterval = TimeSpan.FromSeconds(30); // The very fastest processes inside of Azure Batch accessing anything within pools or jobs appears to use a 30 second polling interval
-        private static readonly TimeSpan shortBackgroundRunInterval = TimeSpan.FromSeconds(1);
-        private static readonly TimeSpan longBackgroundRunInterval = TimeSpan.FromSeconds(2.5);
+        private static readonly TimeSpan shortBackgroundRunInterval = TimeSpan.FromMilliseconds(75);
+        private static readonly TimeSpan longBackgroundRunInterval = TimeSpan.FromSeconds(15);
         private static readonly TimeSpan orphanedTaskInterval = TimeSpan.FromMinutes(10);
         private readonly RunnerEventsProcessor nodeEventProcessor = nodeEventProcessor;
 
@@ -209,9 +209,31 @@ private async ValueTask ProcessQueuedTesTasksAsync(CancellationToken cancellatio
                 .OrderByDescending(t => t.CreationTime)
                 .ToAsyncEnumerable());
 
-            while (!cancellationToken.IsCancellationRequested && (queuedTesTasks.TryDequeue(out var tesTask) || (DateTimeOffset.UtcNow >= nextQueuedRepository && await RefillFromRepository() && queuedTesTasks.TryDequeue(out tesTask))))
+            cancellationToken.ThrowIfCancellationRequested();
+            var now = DateTimeOffset.UtcNow;
+
+            HashSet<TesTask> tasks = new(new TesTasByIdComparer());
+
+            while (queuedTesTasks.TryDequeue(out var tesTask))
             {
-                await ProcessOrchestratedTesTaskAsync("Queued", new(BatchScheduler.ProcessQueuedTesTaskAsync(tesTask, cancellationToken), tesTask), Requeue, cancellationToken);
+                _ = tasks.Add(tesTask);
+            }
+
+            // Catch any tasks reset back to Queued
+            if (nextQueuedRepository <= now)
+            {
+                nextQueuedRepository = now + queuedRepositoryInterval;
+                await (await query(cancellationToken)).ForEachAsync(task => _ = tasks.Add(task), cancellationToken);
+            }
+
+            tasks.ForEach(QueueTesTask);
+
+            void QueueTesTask(TesTask tesTask)
+            {
+                _ = BatchScheduler.ProcessQueuedTesTaskAsync(tesTask, cancellationToken)
+                    .ContinueWith(task => ProcessOrchestratedTesTaskAsync("Queued", new(task, tesTask), Requeue, cancellationToken).AsTask())
+                    .Unwrap()
+                    .ContinueWith(task => Logger.LogError(task.Exception, "Failure to queue TesTask {TesTask}", tesTask.Id), TaskContinuationOptions.OnlyOnFaulted);
             }
 
             async ValueTask Requeue(RepositoryCollisionException<TesTask> exception)
@@ -223,14 +245,6 @@ async ValueTask Requeue(RepositoryCollisionException<TesTask> exception)
                     queuedTesTasks.Enqueue(tesTask);
                 }
             }
-
-            // Catch any tasks reset back to Queued
-            async ValueTask<bool> RefillFromRepository()
-            {
-                nextQueuedRepository = DateTimeOffset.UtcNow + queuedRepositoryInterval;
-                await (await query(cancellationToken)).ForEachAsync(queuedTesTasks.Enqueue, cancellationToken);
-                return !queuedTesTasks.IsEmpty;
-            }
         }
 
         /// <summary>
@@ -538,6 +552,12 @@ await OrchestrateTesTasksOnBatchAsync(
                 cancellationToken);
         }
 
+        private sealed class TesTasByIdComparer : IEqualityComparer<TesTask>
+        {
+            bool IEqualityComparer<TesTask>.Equals(TesTask x, TesTask y) => x?.Id.Equals(y.Id) ?? false;
+            int IEqualityComparer<TesTask>.GetHashCode(TesTask obj) => obj.Id?.GetHashCode() ?? 0;
+        }
+
         /// <inheritdoc/>
         void ITaskScheduler.QueueTesTask(TesTask tesTask)
         {

src/build-push-acr/AcrBuild.cs

@@ -92,8 +92,8 @@ public async ValueTask LoadAsync(IArchive archive, ArmEnvironment environment, C
             {
                 maxTag = await repository.GetAllManifestPropertiesAsync(cancellationToken: cancellationToken)
                     .SelectMany(props => props.Tags.ToAsyncEnumerable())
-                    .Where(tag => tag.StartsWith(this.tag.Version.ToString(3)) && Version.TryParse(tag, out _))
-                    .Select(tag => new Version(tag))
+                    .Where(tag => tag.StartsWith(this.tag.Version.ToString(3)) && Version.TryParse(tag.Contains('-') ? tag[..tag.IndexOf('-')] : tag, out _))
+                    .Select(tag => new Version(tag.Contains('-') ? tag[..tag.IndexOf('-')] : tag))
                     .MaxAsync(cancellationToken);
             }
             catch (Azure.RequestFailedException ex) when (ex.Status == 404)

src/deploy-tes-on-azure/scripts/helm/templates/tes-deployment.yaml

@@ -128,7 +128,7 @@ spec:
             - containerPort: {{ .Values.service.tesPort }}
           resources:
             requests:
-              cpu: "1.5"
+              cpu: "2.5"
               memory: "3072Mi"
             limits:
               cpu: "8.0"