diff --git a/Public/Src/Cache/ContentStore/Distributed/NuCache/ContentLocationDatabase.cs b/Public/Src/Cache/ContentStore/Distributed/NuCache/ContentLocationDatabase.cs
index 68599fb30b..7124ad4757 100644
--- a/Public/Src/Cache/ContentStore/Distributed/NuCache/ContentLocationDatabase.cs
+++ b/Public/Src/Cache/ContentStore/Distributed/NuCache/ContentLocationDatabase.cs
@@ -27,6 +27,7 @@
 using BuildXL.Engine.Cache.KeyValueStores;
 using BuildXL.Utilities;
 using BuildXL.Utilities.Collections;
+using BuildXL.Utilities.ParallelAlgorithms;
 using BuildXL.Utilities.Serialization;
 using BuildXL.Utilities.Tasks;
 using BuildXL.Utilities.Tracing;
diff --git a/Public/Src/Cache/ContentStore/Distributed/NuCache/EventStreaming/ContentLocationEventStore.cs b/Public/Src/Cache/ContentStore/Distributed/NuCache/EventStreaming/ContentLocationEventStore.cs
index a1a314b540..40c1466ff9 100644
--- a/Public/Src/Cache/ContentStore/Distributed/NuCache/EventStreaming/ContentLocationEventStore.cs
+++ b/Public/Src/Cache/ContentStore/Distributed/NuCache/EventStreaming/ContentLocationEventStore.cs
@@ -20,6 +20,7 @@
 using BuildXL.Cache.ContentStore.Utils;
 using BuildXL.Utilities;
 using BuildXL.Utilities.Collections;
+using BuildXL.Utilities.ParallelAlgorithms;
 using BuildXL.Utilities.Tracing;
 using static BuildXL.Cache.ContentStore.Distributed.Tracing.TracingStructuredExtensions;
 using static BuildXL.Cache.ContentStore.Distributed.NuCache.EventStreaming.ContentLocationEventStoreCounters;
diff --git a/Public/Src/Cache/DistributedCache.Host/Configuration/LoggingSettings.cs b/Public/Src/Cache/DistributedCache.Host/Configuration/LoggingSettings.cs
index 893e0f5311..a63fdde9c2 100644
--- a/Public/Src/Cache/DistributedCache.Host/Configuration/LoggingSettings.cs
+++ b/Public/Src/Cache/DistributedCache.Host/Configuration/LoggingSettings.cs
@@ -32,6 +32,12 @@ public class LoggingSettings
         [DataMember]
         public string? MdmAccountName { get; set; }
 
+        /// <summary>
+        /// If true, then the metrics saved asynchronously using a queue.
+        /// </summary>
+        [DataMember]
+        public bool SaveMetricsAsynchronously { get; set; }
+
         public LoggingSettings()
         {
         }
diff --git a/Public/Src/Cache/DistributedCache.Host/Service/LoggerFactory.cs b/Public/Src/Cache/DistributedCache.Host/Service/LoggerFactory.cs
index 51dedc76fd..8d4a35fb0f 100644
--- a/Public/Src/Cache/DistributedCache.Host/Service/LoggerFactory.cs
+++ b/Public/Src/Cache/DistributedCache.Host/Service/LoggerFactory.cs
@@ -83,7 +83,7 @@ public static (ILogger Logger, IDisposable? DisposableToken) ReplaceLogger(Logge
                 else if (!string.IsNullOrEmpty(loggingSettings.MdmAccountName))
                 {
                     Tracer.Debug(context, "Creating MetricsLogger with an in-proc MdmOperationLogger.");
-                    operationLogger = MdmOperationLogger.Create(context, loggingSettings.MdmAccountName, GetDefaultDimensions(arguments));
+                    operationLogger = MdmOperationLogger.Create(context, loggingSettings.MdmAccountName, GetDefaultDimensions(arguments), loggingSettings.SaveMetricsAsynchronously);
                     replacementLogger = new MetricsAdapter(nLogAdapter, operationLogger);
                 }
 
diff --git a/Public/Src/Cache/Logging/Library/BuildXL.Cache.Logging.Library.dsc b/Public/Src/Cache/Logging/Library/BuildXL.Cache.Logging.Library.dsc
index 877e4d3d54..38456d418d 100644
--- a/Public/Src/Cache/Logging/Library/BuildXL.Cache.Logging.Library.dsc
+++ b/Public/Src/Cache/Logging/Library/BuildXL.Cache.Logging.Library.dsc
@@ -10,7 +10,9 @@ namespace Library {
     export const dll = BuildXLSdk.library({
         assemblyName: "BuildXL.Cache.Logging",
         sources: globR(d`.`,"*.cs"),
+        nullable: true,
         references: [
+            BuildXLSdk.asyncInterfacesPackage,
             importFrom("WindowsAzure.Storage").pkg,
             importFrom("NLog").pkg,
             importFrom("BuildXL.Cache.ContentStore").Hashing.dll,
diff --git a/Public/Src/Cache/Logging/Library/External/NLogAdapter.cs b/Public/Src/Cache/Logging/Library/External/NLogAdapter.cs
index ff974a10e6..65d1b82eba 100644
--- a/Public/Src/Cache/Logging/Library/External/NLogAdapter.cs
+++ b/Public/Src/Cache/Logging/Library/External/NLogAdapter.cs
@@ -2,7 +2,6 @@
 // Licensed under the MIT License.
 
 using System;
-using System.Collections.Generic;
 using System.Diagnostics.ContractsLight;
 using System.Threading;
 using System.Threading.Tasks;
@@ -10,6 +9,8 @@
 using NLog;
 using ILogger = BuildXL.Cache.ContentStore.Interfaces.Logging.ILogger;
 
+#nullable enable
+
 namespace BuildXL.Cache.Logging.External
 {
     /// <summary>
@@ -303,19 +304,19 @@ public void Dispose()
             }
         }
 
-        private void CurrentDomain_ProcessExit(object sender, EventArgs e)
+        private void CurrentDomain_ProcessExit(object? sender, EventArgs e)
         {
             Dispose();
         }
 
-        private void CurrentDomain_DomainUnload(object sender, EventArgs e)
+        private void CurrentDomain_DomainUnload(object? sender, EventArgs e)
         {
             Dispose();
         }
 
-        private void CurrentDomain_UnhandledException(object sender, UnhandledExceptionEventArgs args)
+        private void CurrentDomain_UnhandledException(object? sender, UnhandledExceptionEventArgs args)
         {
-            var exception = args.ExceptionObject as Exception;
+            var exception = (Exception)args.ExceptionObject;
 
             string exitString = args.IsTerminating
                                     ? "Process will exit"
@@ -334,7 +335,7 @@ private void CurrentDomain_UnhandledException(object sender, UnhandledExceptionE
             }
         }
 
-        private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs args)
+        private void TaskScheduler_UnobservedTaskException(object? sender, UnobservedTaskExceptionEventArgs args)
         {
             var exception = args.Exception;
             var exceptionObserver = new TaskExceptionObserver();
@@ -367,7 +368,7 @@ private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTask
             }
         }
 
-        private void HandleConfigurationReload(object sender, NLog.Config.LoggingConfigurationReloadedEventArgs e)
+        private void HandleConfigurationReload(object? sender, NLog.Config.LoggingConfigurationReloadedEventArgs e)
         {
             if (e.Succeeded)
             {
@@ -381,7 +382,7 @@ private void HandleConfigurationReload(object sender, NLog.Config.LoggingConfigu
             CurrentSeverity = ComputeCurrentSeverity();
         }
 
-        private void HandleConfigurationChange(object sender, NLog.Config.LoggingConfigurationChangedEventArgs e)
+        private void HandleConfigurationChange(object? sender, NLog.Config.LoggingConfigurationChangedEventArgs e)
         {
             _host.Info("NLog configuration has changed");
 
diff --git a/Public/Src/Cache/Logging/Library/Metrics/FailingLogger.cs b/Public/Src/Cache/Logging/Library/Metrics/FailingLogger.cs
index 441168e872..7fb7ba3d70 100644
--- a/Public/Src/Cache/Logging/Library/Metrics/FailingLogger.cs
+++ b/Public/Src/Cache/Logging/Library/Metrics/FailingLogger.cs
@@ -12,7 +12,7 @@ namespace BuildXL.Cache.Logging
     public abstract class FailingLogger : ILogger
     {
         /// <inheritdoc />
-        public void Dispose()
+        public virtual void Dispose()
         {
             // Throw is the only exception and should not throw.
         }
diff --git a/Public/Src/Cache/Logging/Library/Metrics/MdmOperationLogger.cs b/Public/Src/Cache/Logging/Library/Metrics/MdmOperationLogger.cs
index 16e7e6823f..422ab6449e 100644
--- a/Public/Src/Cache/Logging/Library/Metrics/MdmOperationLogger.cs
+++ b/Public/Src/Cache/Logging/Library/Metrics/MdmOperationLogger.cs
@@ -7,7 +7,6 @@
 using System.Diagnostics.ContractsLight;
 using BuildXL.Cache.ContentStore.Interfaces.Logging;
 using BuildXL.Cache.ContentStore.Interfaces.Tracing;
-using BuildXL.Cache.ContentStore.Tracing.Internal;
 
 #nullable enable
 
@@ -51,7 +50,11 @@ private MdmOperationLogger(Context context, string? monitoringAccount, MetricLog
         /// <remarks>
         /// The mdm metrics are off if <paramref name="monitoringAccount"/> is null or empty.
         /// </remarks>
-        public static MdmOperationLogger Create(Context context, string? monitoringAccount, List<DefaultDimension> defaultDimensions)
+        public static MdmOperationLogger Create(
+            Context context,
+            string? monitoringAccount,
+            List<DefaultDimension> defaultDimensions,
+            bool saveMetricsAsynchronously)
         {
             // Setting the default dimensions once instead of passing them all the time explicitly.
             MetricLogger.InitializeMdmDefaultDimensions(context, defaultDimensions);
@@ -63,7 +66,8 @@ public static MdmOperationLogger Create(Context context, string? monitoringAccou
                 logicalNameSpace: ServiceName,
                 metricName: "OperationDurationMs",
                 addDefaultDimensions: true,
-                dimensions: OperationDurationDimensions);
+                dimensions: OperationDurationDimensions,
+                saveMetricsAsynchronously);
 
             var metricLogger = MetricLogger.CreateLogger(
                 context,
@@ -71,22 +75,30 @@ public static MdmOperationLogger Create(Context context, string? monitoringAccou
                 logicalNameSpace: ServiceName,
                 metricName: "Metric",
                 addDefaultDimensions: true,
-                dimensions: MetricDimensions);
+                dimensions: MetricDimensions,
+                saveMetricsAsynchronously);
 
             return new MdmOperationLogger(context, monitoringAccount, operationFinishedMetricLogger, metricLogger);
         }
 
+        /// <inheritdoc />
+        public override void Dispose()
+        {
+            _metricLogger.Dispose();
+        }
+
         /// <inheritdoc />
         public void OperationFinished(in OperationResult result)
         {
             _operationFinishedMetricLogger.Log(
+                new OperationFinishedMetric(
                 (long)result.Duration.TotalMilliseconds,
                 result.OperationName,
                 result.OperationKind.ToStringNoAlloc(),
                 result.Status == OperationStatus.Success ? "Succeeded" : "Failed",
                 result.Status.ToStringNoAlloc(),
                 result.TracerName,
-                result.Exception?.GetType().ToString() ?? "NA");
+                    result.Exception?.GetType().ToString() ?? "NA"));
         }
 
         /// <inheritdoc />
@@ -113,7 +125,8 @@ MetricLogger getOrCreateMetricLogger(string statisticName)
                         logicalNameSpace: ServiceName,
                         metricName: statisticName,
                         addDefaultDimensions: true,
-                        dimensions: Array.Empty<Dimension>());
+                        dimensions: Array.Empty<Dimension>(),
+                        saveMetricsAsynchronously: false); // we should not have too many stats metrics, so its fine to save them synchronously all the time.
 
                     _statisticLoggers[statisticName] = metricLogger;
                 }
diff --git a/Public/Src/Cache/Logging/Library/Metrics/MetricLogger.cs b/Public/Src/Cache/Logging/Library/Metrics/MetricLogger.cs
index e745f3008b..e60b5f89ad 100644
--- a/Public/Src/Cache/Logging/Library/Metrics/MetricLogger.cs
+++ b/Public/Src/Cache/Logging/Library/Metrics/MetricLogger.cs
@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System;
 using System.Collections.Generic;
 using BuildXL.Cache.ContentStore.Interfaces.Tracing;
 
@@ -8,10 +9,22 @@
 
 namespace BuildXL.Cache.Logging
 {
+    /// <summary>
+    /// Represents a metric emitted when an operation is finished.
+    /// </summary>
+    public record struct OperationFinishedMetric(
+        long DurationMs,
+        string OperationName,
+        string OperationKind,
+        string SuccessOrFailure,
+        string Status,
+        string Component,
+        string ExceptionType);
+
     /// <summary>
     /// A base type for logging metrics.
     /// </summary>
-    public abstract class MetricLogger
+    public abstract class MetricLogger : IDisposable
     {
         /// <summary>
         /// Logs metric value with the given dimensions.
@@ -20,11 +33,21 @@ public abstract class MetricLogger
         /// <param name="dimensionValues">Dimension values.</param>
         public abstract void Log(long metricValue, params string?[] dimensionValues);
 
+        /// <summary>
+        /// Logs a metric of a finished operation.
+        /// </summary>
+        public abstract void Log(in OperationFinishedMetric metric);
+
         /// <summary>
         /// Logs metric value with the given dimensions.
         /// </summary>
         public abstract void Log(long metricValue, string dimension1, string dimension2);
 
+        /// <summary>
+        /// Dispose remaining logging resources or flush the logs.
+        /// </summary>
+        public virtual void Dispose() { }
+
         /// <summary>
         /// Geneva / MDM is not tolerant of null or empty dimension values. Replace them with actual strings to prevent errors.
         /// </summary>
@@ -68,10 +91,11 @@ public static MetricLogger CreateLogger(
             string logicalNameSpace,
             string metricName,
             bool addDefaultDimensions,
-            IEnumerable<Dimension> dimensions)
+            IEnumerable<Dimension> dimensions,
+            bool saveMetricsAsynchronously)
         {
 #if MICROSOFT_INTERNAL
-            return WindowsMetricLogger.Create(context, monitoringAccount, logicalNameSpace, metricName, addDefaultDimensions, dimensions);
+            return WindowsMetricLogger.Create(context, monitoringAccount, logicalNameSpace, metricName, addDefaultDimensions, dimensions, saveMetricsAsynchronously);
 #else
             return NoOpMetricLogger.Instance;
 #endif
diff --git a/Public/Src/Cache/Logging/Library/Metrics/NoOpMetricLogger.cs b/Public/Src/Cache/Logging/Library/Metrics/NoOpMetricLogger.cs
index 3b1d694c8f..c54b141623 100644
--- a/Public/Src/Cache/Logging/Library/Metrics/NoOpMetricLogger.cs
+++ b/Public/Src/Cache/Logging/Library/Metrics/NoOpMetricLogger.cs
@@ -18,6 +18,10 @@ internal sealed class NoOpMetricLogger : MetricLogger
         /// <inheritdoc />
         public override void Log(long metricValue, params string?[] dimensionValues) { }
 
+        /// <inheritdoc />
         public override void Log(long metricValue, string dimension1, string dimension2) { }
+
+        /// <inheritdoc />
+        public override void Log(in OperationFinishedMetric metric) { }
     }
 }
diff --git a/Public/Src/Cache/Logging/Library/Metrics/WindowsMetricsLogger.cs b/Public/Src/Cache/Logging/Library/Metrics/WindowsMetricsLogger.cs
index 1a0e0f1248..e4eb8ad7ec 100644
--- a/Public/Src/Cache/Logging/Library/Metrics/WindowsMetricsLogger.cs
+++ b/Public/Src/Cache/Logging/Library/Metrics/WindowsMetricsLogger.cs
@@ -4,28 +4,41 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Diagnostics.ContractsLight;
 using System.Linq;
-using BuildXL.Cache.ContentStore.Hashing;
+using System.Threading.Tasks;
 using BuildXL.Cache.ContentStore.Interfaces.Tracing;
 using BuildXL.Cache.ContentStore.Tracing;
+using BuildXL.Utilities.Collections;
+using BuildXL.Utilities.ParallelAlgorithms;
 using Microsoft.Cloud.InstrumentationFramework;
 
 #nullable enable
 
 namespace BuildXL.Cache.Logging
 {
+    /// <nodoc />
+    internal readonly record struct SaveMetricInput(long Metric, string[] Dimensions, bool ReturnDimensionsToPool);
+
     /// <summary>
     /// Implements metric logger on top of the Ifx MeasureMetric.  Logs to MDM.
     /// </summary>
-    internal sealed class WindowsMetricLogger : MetricLogger
+    internal class WindowsMetricLogger : MetricLogger
     {
-        private static readonly Tracer Tracer = new Tracer(nameof(WindowsMetricLogger));
-
+        private const int OperationFinishedDimensionsLength = 6;
         /// <summary>
         /// A pool for dimension arrays. Using 'Pool' and not 'ArrayPool' because this is lighter weight and we always need to have the exact number of elements.
         /// </summary>
-        private static readonly Pool<string[]> DimensionsArrayPool = new Pool<string[]>(factory: () => new string[2]);
+        private static readonly Utilities.ObjectPool<string[]> DimensionsArrayPool = new (creator: () => new string[2], cleanup: _ => { });
+        private static readonly Utilities.ObjectPool<string[]> OperationFinishedDimensionsArrayPool = new (creator: () => new string[OperationFinishedDimensionsLength], cleanup: _ => { });
+
+        private static readonly Tracer Tracer = new Tracer(nameof(WindowsMetricLogger));
+
+        private readonly INagleQueue<SaveMetricInput>? _metricsSavingQueue;
+
+        [MemberNotNullWhen(true, nameof(_metricsSavingQueue))]
+        private bool SaveMetricsAsynchronously { get; }
 
         private readonly Context _context;
         private readonly string _logicalNameSpace;
@@ -35,12 +48,13 @@ internal sealed class WindowsMetricLogger : MetricLogger
         private readonly MeasureMetric0D? _measureMetric0D;
 
         /// <nodoc />
-        private WindowsMetricLogger(
+        protected WindowsMetricLogger(
             Context context,
             string logicalNameSpace,
             string metricName,
             MeasureMetric? measureMetric,
-            MeasureMetric0D? measureMetric0D)
+            MeasureMetric0D? measureMetric0D,
+            bool saveMetricsAsynchronously)
         {
             Contract.Requires(measureMetric is not null || measureMetric0D is not null);
 
@@ -50,6 +64,25 @@ private WindowsMetricLogger(
 
             _measureMetric = measureMetric;
             _measureMetric0D = measureMetric0D;
+
+            SaveMetricsAsynchronously = saveMetricsAsynchronously;
+            if (saveMetricsAsynchronously)
+            {
+                _metricsSavingQueue = NagleQueueFactory.Create<SaveMetricInput>(
+                    metrics =>
+                    {
+                        foreach (var (metric, dimensions, returnToPool) in metrics.AsStructEnumerable())
+                        {
+                            LogCore(metric, dimensions, returnToPool);
+                        }
+
+                        return Task.CompletedTask;
+                    },
+                    maxDegreeOfParallelism: 1,
+                    interval: TimeSpan.FromSeconds(1),
+                    batchSize: 1000);
+            }
+            
         }
 
         /// <summary>
@@ -65,7 +98,8 @@ public static MetricLogger Create(
             string logicalNameSpace,
             string metricName,
             bool addDefaultDimensions,
-            IEnumerable<Dimension> dimensions)
+            IEnumerable<Dimension> dimensions,
+            bool saveMetricsAsynchronously)
         {
             if (string.IsNullOrEmpty(monitoringAccount))
             {
@@ -75,7 +109,7 @@ public static MetricLogger Create(
             // For some reason the generic MeasureMetric class does not work with 0 dimensions.  There is a special
             // class (MeasureMetric0D) that does metrics without any dimensions.
             var dimensionNames = dimensions.Select(d => d.Name).ToArray();
-            Tracer.Debug(context, $"Initializing Mdm logger {logicalNameSpace}:{metricName}");
+            Tracer.Debug(context, $"Initializing Mdm logger {logicalNameSpace}:{metricName}. AsyncLogging={saveMetricsAsynchronously}.");
             var error = new ErrorContext();
 
             MeasureMetric? measureMetric = null;
@@ -97,7 +131,7 @@ public static MetricLogger Create(
                 return NoOpMetricLogger.Instance;
             }
 
-            return new WindowsMetricLogger(context, logicalNameSpace, metricName, measureMetric, measureMetric0D);
+            return new WindowsMetricLogger(context, logicalNameSpace, metricName, measureMetric, measureMetric0D, saveMetricsAsynchronously);
         }
 
         /// <summary>
@@ -136,26 +170,78 @@ public static void InitializeDefaultDimensions(Context context, List<DefaultDime
             }
         }
 
+        /// <inheritdoc />
+        public override void Dispose()
+        {
+            if (SaveMetricsAsynchronously)
+            {
+                _metricsSavingQueue.Dispose();
+            }
+        }
+
         /// <inheritdoc />
         public override void Log(long metricValue, params string?[] dimensionValues)
         {
             ReplaceNullAndEmpty(dimensionValues);
 
-            LogCore(metricValue, dimensionValues!);
+            // We don't control an incoming array, so we can't pool it.
+            LogMetric(metricValue, dimensionValues!, returnToPool: false);
+        }
+
+        private void LogMetric(long metricValue, string[] dimensionValues, bool returnToPool)
+        {
+            if (SaveMetricsAsynchronously)
+            {
+                _metricsSavingQueue.Enqueue(new SaveMetricInput(metricValue, dimensionValues, ReturnDimensionsToPool: returnToPool));
+            }
+            else
+            {
+                LogCore(metricValue, dimensionValues!, returnToPool: returnToPool);
+            }
         }
 
         /// <inheritdoc />
         public override void Log(long metricValue, string dimension1, string dimension2)
         {
-            using var handle = DimensionsArrayPool.Get();
+            var dimensions = DimensionsArrayPool.GetInstance().Instance;
 
-            var dimensions = handle.Value;
             dimensions[0] = dimension1;
             dimensions[1] = dimension2;
-            LogCore(metricValue, dimensions);
+
+            LogMetric(metricValue, dimensions, returnToPool: true);
         }
 
-        private void LogCore(long metricValue, string[] dimensionValues)
+        /// <inheritdoc />
+        public override void Log(in OperationFinishedMetric metric)
+        {
+            var dimensions = OperationFinishedDimensionsArrayPool.GetInstance().Instance;
+            int i = 0;
+            dimensions[i++] = metric.OperationName;
+            dimensions[i++] = metric.OperationKind;
+            dimensions[i++] = metric.SuccessOrFailure;
+            dimensions[i++] = metric.Status;
+            dimensions[i++] = metric.Component;
+            dimensions[i++] = metric.ExceptionType;
+            var metricValue = metric.DurationMs;
+
+            LogMetric(metricValue, dimensions, returnToPool: true);
+        }
+
+        /// <summary>
+        /// Returns a given <paramref name="dimensionValues"/> array to a corresponding array pool based on the array's length.
+        /// </summary>
+        protected void ReturnToPool(string[] dimensionValues)
+        {
+            var pool = dimensionValues.Length == OperationFinishedDimensionsLength
+                ? OperationFinishedDimensionsArrayPool
+                : DimensionsArrayPool;
+            pool.PutInstance(dimensionValues);
+        }
+
+        /// <summary>
+        /// Log the metric and the dimensions.
+        /// </summary>
+        protected virtual void LogCore(long metricValue, string[] dimensionValues, bool returnToPool)
         {
             var error = new ErrorContext();
 
@@ -168,6 +254,11 @@ private void LogCore(long metricValue, string[] dimensionValues)
                 Tracer.Error(_context,
                     $"Fail to log metric value. MetricValue='{metricValue}', DimensionValues='{string.Join(", ", dimensionValues)}', {_logicalNameSpace}:{_metricName} ErrorCode: {error.ErrorCode} ErrorMessage: {error.ErrorMessage}");
             }
+
+            if (returnToPool)
+            {
+                ReturnToPool(dimensionValues);
+            }
         }
     }
 }
diff --git a/Public/Src/Cache/Logging/Test/BuildXL.Cache.Logging.Test.dsc b/Public/Src/Cache/Logging/Test/BuildXL.Cache.Logging.Test.dsc
index 51e33fa70d..27ea003eb8 100644
--- a/Public/Src/Cache/Logging/Test/BuildXL.Cache.Logging.Test.dsc
+++ b/Public/Src/Cache/Logging/Test/BuildXL.Cache.Logging.Test.dsc
@@ -18,6 +18,7 @@ namespace Test {
             importFrom("BuildXL.Cache.ContentStore").Library.dll,
             importFrom("BuildXL.Cache.ContentStore").InterfacesTest.dll,
             importFrom("BuildXL.Cache.ContentStore").Test.dll,
+            importFrom("BuildXL.Utilities").dll,
             ...BuildXLSdk.bclAsyncPackages,
             
             ...addIfLazy(BuildXLSdk.Flags.isMicrosoftInternal, () => [
diff --git a/Public/Src/Cache/Logging/Test/MdmOperationLoggerTests.cs b/Public/Src/Cache/Logging/Test/MdmOperationLoggerTests.cs
index 6999c11f72..7ff508938c 100644
--- a/Public/Src/Cache/Logging/Test/MdmOperationLoggerTests.cs
+++ b/Public/Src/Cache/Logging/Test/MdmOperationLoggerTests.cs
@@ -3,10 +3,17 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
+using System.Text;
 using BuildXL.Cache.ContentStore.Interfaces.Logging;
 using BuildXL.Cache.ContentStore.Interfaces.Tracing;
 using BuildXL.Cache.ContentStore.Logging;
+using BuildXL.Utilities;
+using Microsoft.Cloud.InstrumentationFramework;
 using Xunit;
+using OperationResult = BuildXL.Cache.ContentStore.Interfaces.Logging.OperationResult;
+
+#nullable enable
 
 namespace BuildXL.Cache.Logging.Test
 {
@@ -19,8 +26,210 @@ public void MetricOperationsAreNotFailing()
             // but we can check that all the required dependencies are available.
             // And this test just checks that all the managed and native dlls are presented and all the types can be loaded successfully.
             var context = new Context(new Logger());
-            var logger = MdmOperationLogger.Create(context, "CloudBuildCBTest", new List<DefaultDimension>());
+            using var logger = MdmOperationLogger.Create(context, "CloudBuildCBTest", new List<DefaultDimension>(), saveMetricsAsynchronously: false);
+            logger.OperationFinished(new OperationResult("CustomMessage", "Test OperationName", "TestTracer", OperationStatus.Success, TimeSpan.FromSeconds(42), OperationKind.None, exception: null, operationId: "42", severity: Severity.Debug));
+        }
+
+        [Fact]
+        public void MetricOperationsAreNotFailingWhenSavingAsynchronously()
+        {
+            var context = new Context(new Logger());
+            using var logger = MdmOperationLogger.Create(context, "CloudBuildCBTest", new List<DefaultDimension>(), saveMetricsAsynchronously: true);
             logger.OperationFinished(new OperationResult("CustomMessage", "Test OperationName", "TestTracer", OperationStatus.Success, TimeSpan.FromSeconds(42), OperationKind.None, exception: null, operationId: "42", severity: Severity.Debug));
         }
+
+        [Fact]
+        public void TestAsyncOperationFinished()
+        {
+            var logger = CreateLogger(new Context(new Logger()));
+
+            var inputMetrics = Enumerable.Range(1, 5_005).Select(
+                    v => new OperationFinishedMetric(
+                        v,
+                        $"OperationName{v}",
+                        $"OperationKind{v}",
+                        $"SuccessOrFailure{v}",
+                        $"Status{v}",
+                        $"Component{v}",
+                        $"ExceptionType{v}"))
+                .ToArray();
+            int currentThreadId = Environment.CurrentManagedThreadId;
+            int onLogCoreThreadId = -1;
+            var savedMetrics = new List<OperationFinishedMetric>();
+            logger.OnLogCore = tuple =>
+                               {
+                                   onLogCoreThreadId = Environment.CurrentManagedThreadId;
+                                   var dimensions = tuple.dimensions;
+                                   savedMetrics.Add(new OperationFinishedMetric(tuple.metricValue,
+                                       dimensions[0],
+                                       dimensions[1],
+                                       dimensions[2],
+                                       dimensions[3],
+                                       dimensions[4],
+                                       dimensions[5]));
+                               };
+
+            foreach (var input in inputMetrics)
+            {
+                logger.Log(input);
+            }
+
+            logger.Dispose();
+
+            Assert.Equal(inputMetrics, savedMetrics);
+            Assert.NotEqual(currentThreadId, onLogCoreThreadId);
+        }
+
+        [Fact]
+        public void TestAsyncLog2DimensionsStress()
+        {
+            int i = 0;
+            try
+            {
+                for (i = 0; i < 1000; i++)
+                {
+                    TestAsyncLog2Dimensions();
+                }
+            }
+            catch (Exception e)
+            {
+                throw new Exception($"I: {i}", e);
+            }
+        }
+
+        [Fact]
+        public void TestAsyncLog2Dimensions()
+        {
+            var logger = CreateLogger(new Context(new Logger()));
+
+            var inputMetrics = Enumerable.Range(1, 5_005).Select(v => new TestMetric(v, Dimensions.Generate(dimensionLength: 2, v))).ToArray();
+
+            int currentThreadId = Environment.CurrentManagedThreadId;
+            int onLogCoreThreadId = -1;
+
+            var savedMetrics = new List<TestMetric>();
+            logger.OnLogCore = tuple =>
+                               {
+                                   onLogCoreThreadId = Environment.CurrentManagedThreadId;
+                                   // Need to make a copy of 'dimensions' array because the array is pooled.
+                                   savedMetrics.Add(new TestMetric(tuple.metricValue, new Dimensions(tuple.dimensions.ToArray())));
+                               };
+
+            foreach (var input in inputMetrics)
+            {
+                logger.Log(input.Value, input.Dimensions.DimensionValues[0], input.Dimensions.DimensionValues[1]);
+            }
+
+            logger.Dispose();
+
+            Assert.Equal(savedMetrics.ToArray(), inputMetrics);
+            Assert.NotEqual(currentThreadId, onLogCoreThreadId);
+        }
+
+        [InlineData(1)]
+        [InlineData(2)]
+        [InlineData(5)]
+        [InlineData(15)]
+        [Theory]
+        public void TestAsyncLogDimensions(int dimensionLength)
+        {
+            var logger = CreateLogger(new Context(new Logger()));
+
+            var inputMetrics = Enumerable.Range(1, 5_005).Select(v => new TestMetric(v, Dimensions.Generate(dimensionLength, v))).ToArray();
+
+            int currentThreadId = Environment.CurrentManagedThreadId;
+            int onLogCoreThreadId = -1;
+
+            var savedMetrics = new List<TestMetric>();
+            logger.OnLogCore = tuple =>
+                               {
+                                   onLogCoreThreadId = Environment.CurrentManagedThreadId;
+                                   savedMetrics.Add(new TestMetric(tuple.metricValue, new Dimensions(tuple.dimensions)));
+                               };
+
+            foreach (var input in inputMetrics)
+            {
+                logger.Log(input.Value, input.Dimensions.DimensionValues);
+            }
+
+            logger.Dispose();
+
+            Assert.Equal(inputMetrics, savedMetrics);
+            Assert.NotEqual(currentThreadId, onLogCoreThreadId);
+        }
+
+        private static WindowsMetricLoggerTester CreateLogger(Context context)
+        {
+            var error = new ErrorContext();
+            string logicalNameSpace = "logicalNameSpace";
+            string metricName = "metricName";
+            var measureMetric = MeasureMetric.Create("monitoringAccount", logicalNameSpace, metricName, ref error, addDefaultDimension: true, Array.Empty<string>());
+
+            return new WindowsMetricLoggerTester(
+                context,
+                logicalNameSpace,
+                metricName,
+                measureMetric: measureMetric,
+                measureMetric0D: null,
+                saveMetricsAsynchronously: true);
+        }
+
+        private class WindowsMetricLoggerTester : WindowsMetricLogger
+        {
+            public Action<(long metricValue, string[] dimensions)>? OnLogCore;
+
+            public WindowsMetricLoggerTester(Context context, string logicalNameSpace, string metricName, MeasureMetric? measureMetric, MeasureMetric0D? measureMetric0D, bool saveMetricsAsynchronously)
+                : base(context, logicalNameSpace, metricName, measureMetric, measureMetric0D, saveMetricsAsynchronously)
+            {
+            }
+
+            /// <inheritdoc />
+            protected override void LogCore(long metricValue, string[] dimensionValues, bool returnToPool)
+            {
+                OnLogCore?.Invoke((metricValue, dimensionValues));
+
+                if (returnToPool)
+                {
+                    ReturnToPool(dimensionValues);
+                }
+            }
+        }
+
+        private readonly record struct Dimensions(string[] DimensionValues)
+        {
+            /// <inheritdoc />
+            public bool Equals(Dimensions other)
+            {
+                return DimensionValues.SequenceEqual(other.DimensionValues);
+            }
+
+            /// <inheritdoc />
+            public override int GetHashCode()
+            {
+                int hashCode = 42;
+                foreach (var d in DimensionValues)
+                {
+                    hashCode = HashCodeHelper.Combine(hashCode, d.GetHashCode());
+                }
+
+                return hashCode;
+            }
+
+            public static Dimensions Generate(int dimensionLength, int seed)
+            {
+                return new Dimensions(
+                    Enumerable.Range(1, dimensionLength)
+                        .Select(d => $"D{d}_{seed}")
+                        .ToArray());
+            }
+
+            private bool PrintMembers(StringBuilder builder)
+            {
+                builder.Append(string.Join(", ", DimensionValues));
+                return true;
+            }
+        }
+
+        private record struct TestMetric(long Value, Dimensions Dimensions);
     }
 }
diff --git a/Public/Src/Utilities/Utilities/BuildXL.Utilities.dsc b/Public/Src/Utilities/Utilities/BuildXL.Utilities.dsc
index 67a2c9016f..61e4b15234 100644
--- a/Public/Src/Utilities/Utilities/BuildXL.Utilities.dsc
+++ b/Public/Src/Utilities/Utilities/BuildXL.Utilities.dsc
@@ -39,6 +39,7 @@ export const dll = BuildXLSdk.library({
             })
         ]),
         ...BuildXLSdk.tplPackages,
+        BuildXLSdk.asyncInterfacesPackage,
         importFrom("Newtonsoft.Json").pkg,
         BuildXLSdk.isDotNetCoreApp
             ? importFrom("Grpc.Core.Api").withQualifier({ targetFramework: "netstandard2.1" }).pkg
diff --git a/Public/Src/Cache/ContentStore/Distributed/NuCache/NagleQueueV2.cs b/Public/Src/Utilities/Utilities/ParallelAlgorithms/NagleQueueV2.cs
similarity index 95%
rename from Public/Src/Cache/ContentStore/Distributed/NuCache/NagleQueueV2.cs
rename to Public/Src/Utilities/Utilities/ParallelAlgorithms/NagleQueueV2.cs
index 24a4e8ab9f..cbe4cfc743 100644
--- a/Public/Src/Cache/ContentStore/Distributed/NuCache/NagleQueueV2.cs
+++ b/Public/Src/Utilities/Utilities/ParallelAlgorithms/NagleQueueV2.cs
@@ -7,13 +7,11 @@
 using System.Diagnostics.ContractsLight;
 using System.Threading;
 using System.Threading.Tasks;
-using BuildXL.Utilities;
 using BuildXL.Utilities.Collections;
-using BuildXL.Utilities.ParallelAlgorithms;
 
 #nullable enable
 
-namespace BuildXL.Cache.ContentStore.Distributed.NuCache
+namespace BuildXL.Utilities.ParallelAlgorithms
 {
     /// <nodoc />
     public static class NagleQueueFactory
@@ -132,6 +130,7 @@ public void Enqueue(T item)
             }
         }
 
+        /// <nodoc />
         protected virtual void EnqueueCore(T item)
         {
             List<T>? itemsToProcess = null;