diff --git a/MSBuildCache.sln b/MSBuildCache.sln index 91f9157..85d9f83 100644 --- a/MSBuildCache.sln +++ b/MSBuildCache.sln @@ -34,6 +34,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.MSBuildCache.Loca EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.MSBuildCache.Repack.Tests", "src\Repack.Tests\Microsoft.MSBuildCache.Repack.Tests.csproj", "{3BCB6452-B087-4A03-8418-C79F2715DDE7}" EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.MSBuildCache.AzurePipelines.Tests", "src\AzurePipelines.Tests\Microsoft.MSBuildCache.AzurePipelines.Tests.csproj", "{61A86AEA-F043-4CC4-B60B-A040C5C36194}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|x64 = Debug|x64 @@ -68,6 +70,10 @@ Global {3BCB6452-B087-4A03-8418-C79F2715DDE7}.Debug|x64.Build.0 = Debug|x64 {3BCB6452-B087-4A03-8418-C79F2715DDE7}.Release|x64.ActiveCfg = Release|x64 {3BCB6452-B087-4A03-8418-C79F2715DDE7}.Release|x64.Build.0 = Release|x64 + {61A86AEA-F043-4CC4-B60B-A040C5C36194}.Debug|x64.ActiveCfg = Debug|x64 + {61A86AEA-F043-4CC4-B60B-A040C5C36194}.Debug|x64.Build.0 = Debug|x64 + {61A86AEA-F043-4CC4-B60B-A040C5C36194}.Release|x64.ActiveCfg = Release|x64 + {61A86AEA-F043-4CC4-B60B-A040C5C36194}.Release|x64.Build.0 = Release|x64 EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -80,6 +86,7 @@ Global {97357681-C75E-445D-8547-46F312D01CED} = {EFFB5949-347C-4F28-8964-571D5C6B6209} {F6586428-E047-42C8-B0AC-048DF6DFAF18} = {EFFB5949-347C-4F28-8964-571D5C6B6209} {3BCB6452-B087-4A03-8418-C79F2715DDE7} = {EFFB5949-347C-4F28-8964-571D5C6B6209} + {61A86AEA-F043-4CC4-B60B-A040C5C36194} = {EFFB5949-347C-4F28-8964-571D5C6B6209} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {F1CDA78F-A666-431B-BF44-56DA7DF193BA} diff --git a/src/AzurePipelines.Tests/Microsoft.MSBuildCache.AzurePipelines.Tests.csproj b/src/AzurePipelines.Tests/Microsoft.MSBuildCache.AzurePipelines.Tests.csproj new file mode 100644 index 0000000..610a439 --- /dev/null +++ b/src/AzurePipelines.Tests/Microsoft.MSBuildCache.AzurePipelines.Tests.csproj @@ -0,0 +1,14 @@ + + + + x64 + $(Platform) + net9.0 + Microsoft.MSBuildCache.AzurePipelines.Tests + + $(NoWarn);CA1515 + + + + + diff --git a/src/AzurePipelines.Tests/SelectorPublicationTests.cs b/src/AzurePipelines.Tests/SelectorPublicationTests.cs new file mode 100644 index 0000000..cfafb50 --- /dev/null +++ b/src/AzurePipelines.Tests/SelectorPublicationTests.cs @@ -0,0 +1,162 @@ +// Copyright (c) Microsoft. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using BuildXL.Cache.ContentStore.Hashing; +using BuildXL.Cache.MemoizationStore.Interfaces.Sessions; +using Microsoft.VisualStudio.TestTools.UnitTesting; + +namespace Microsoft.MSBuildCache.AzurePipelines.Tests; + +[TestClass] +public class SelectorPublicationTests +{ + [TestMethod] + public async Task ConcurrentPublicationsMergeTheConflictWinner() + { + Selector existingSelector = CreateSelector(1); + Selector firstSelector = CreateSelector(2); + Selector secondSelector = CreateSelector(3); + Fingerprint weakFingerprint = new("01"); + const string Universe = "universe"; + string latestKey = PipelineCachingCacheClient.ComputeSelectorsReadKey(Universe, weakFingerprint); + PublicationState initial = new("initial", new[] { existingSelector }); + ConcurrentDictionary entries = new(); + ConcurrentQueue conflictQueries = new(); + PublicationState? firstWinner = null; + PublicationState? finalWinner = null; + string? finalWriteKey = null; + int initialAttempts = 0; + TaskCompletionSource bothInitialAttemptsStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource firstWinnerReady = new(TaskCreationOptions.RunContinuationsAsynchronously); + + Task QueryAsync(string key, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + if (key == latestKey) + { + return Task.FromResult(initial); + } + + entries.TryGetValue(key, out PublicationState? entry); + return Task.FromResult(entry); + } + + Task GetConflictWinnerAsync(string key, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + conflictQueries.Enqueue(key); + Assert.IsTrue(entries.TryGetValue(key, out PublicationState? entry)); + return Task.FromResult(entry); + } + + async Task TryPublishAsync( + PublicationState? predecessor, + IReadOnlyCollection selectors, + string key, + CancellationToken cancellationToken) + { + Assert.IsNotNull(predecessor); + + if (predecessor.Id == initial.Id) + { + if (Interlocked.Increment(ref initialAttempts) == 2) + { + bothInitialAttemptsStarted.SetResult(true); + } + + await bothInitialAttemptsStarted.Task; + + if (selectors.Contains(firstSelector)) + { + firstWinner = new PublicationState("first", selectors); + Assert.IsTrue(entries.TryAdd(key, firstWinner)); + firstWinnerReady.SetResult(true); + return true; + } + + await firstWinnerReady.Task; + return false; + } + + Assert.AreEqual(firstWinner!.Id, predecessor.Id); + finalWinner = new PublicationState("final", selectors); + finalWriteKey = key; + Assert.IsTrue(entries.TryAdd(key, finalWinner)); + return true; + } + + Task firstPublication = SelectorPublication.AddAsync( + firstSelector, + latestKey, + QueryAsync, + GetConflictWinnerAsync, + (state, _) => Task.FromResult>(state!.Selectors), + state => PipelineCachingCacheClient.ComputeSelectorsWriteKey(Universe, weakFingerprint, state?.Id), + TryPublishAsync, + CancellationToken.None); + Task secondPublication = SelectorPublication.AddAsync( + secondSelector, + latestKey, + QueryAsync, + GetConflictWinnerAsync, + (state, _) => Task.FromResult>(state!.Selectors), + state => PipelineCachingCacheClient.ComputeSelectorsWriteKey(Universe, weakFingerprint, state?.Id), + TryPublishAsync, + CancellationToken.None); + + Assert.IsTrue(await firstPublication); + Assert.IsTrue(await secondPublication); + CollectionAssert.AreEquivalent( + new[] { existingSelector, firstSelector, secondSelector }, + finalWinner!.Selectors.ToArray()); + string initialWriteKey = PipelineCachingCacheClient.ComputeSelectorsWriteKey(Universe, weakFingerprint, initial.Id); + CollectionAssert.AreEqual(new[] { initialWriteKey }, conflictQueries.ToArray()); + Assert.AreEqual( + PipelineCachingCacheClient.ComputeSelectorsWriteKey(Universe, weakFingerprint, firstWinner!.Id), + finalWriteKey); + StringAssert.StartsWith(latestKey, "selector6|", StringComparison.Ordinal); + } + + [TestMethod] + public void OutputKeyIncludesSelectorOutput() + { + Selector firstSelector = CreateSelector(1); + Selector secondSelector = CreateSelector(2); + Fingerprint weakFingerprint = new("01"); + StrongFingerprint first = new(weakFingerprint, firstSelector); + StrongFingerprint second = new(weakFingerprint, secondSelector); + + string firstKey = PipelineCachingCacheClient.ComputeOutputKey("universe", first, forWrite: false, writeId: 0); + string secondKey = PipelineCachingCacheClient.ComputeOutputKey("universe", second, forWrite: false, writeId: 0); + + Assert.AreNotEqual(firstKey, secondKey); + StringAssert.StartsWith(firstKey, "outputs6|", StringComparison.Ordinal); + StringAssert.Contains(firstKey, "|01|", StringComparison.Ordinal); + StringAssert.Contains(secondKey, "|02|", StringComparison.Ordinal); + } + + private static Selector CreateSelector(byte output) + { + byte[] hashBytes = Enumerable.Repeat((byte)0x5a, 32).ToArray(); + return new Selector(new ContentHash(HashType.SHA256, hashBytes), new[] { output }); + } + + private sealed class PublicationState + { + public PublicationState(string id, IEnumerable selectors) + { + Id = id; + Selectors = new HashSet(selectors); + } + + public string Id { get; } + + public HashSet Selectors { get; } + } +} diff --git a/src/AzurePipelines/Microsoft.MSBuildCache.AzurePipelines.csproj b/src/AzurePipelines/Microsoft.MSBuildCache.AzurePipelines.csproj index b579fbe..fc8751a 100644 --- a/src/AzurePipelines/Microsoft.MSBuildCache.AzurePipelines.csproj +++ b/src/AzurePipelines/Microsoft.MSBuildCache.AzurePipelines.csproj @@ -8,6 +8,9 @@ + + + diff --git a/src/AzurePipelines/PipelineCachingCacheClient.cs b/src/AzurePipelines/PipelineCachingCacheClient.cs index ee12478..0ad0f87 100644 --- a/src/AzurePipelines/PipelineCachingCacheClient.cs +++ b/src/AzurePipelines/PipelineCachingCacheClient.cs @@ -81,7 +81,7 @@ internal sealed class PipelineCachingCacheClient : CacheClient private static readonly string DomainId = WellKnownDomainIds.DefaultDomainId.ToString(); private const char KeySegmentSeperator = '|'; - private const int InternalSeed = 5; + private const int InternalSeed = 6; private readonly bool _remoteCacheIsReadOnly; private readonly string _universe; private readonly IAppTraceSource _azureDevopsTracer; @@ -323,7 +323,37 @@ protected override async Task AddNodeAsync( } // add the WFP -> Selector mapping - bool wfpAddded; + await _startupTask; + string selectorsReadKey = ComputeSelectorsReadKey(_universe, fingerprint.WeakFingerprint); + bool wfpAddded = await SelectorPublication.AddAsync( + fingerprint.Selector, + selectorsReadKey, + (key, ct) => QueryPipelineCaching( + context, + new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(key.Split(KeySegmentSeperator)), + ct), + (key, ct) => QueryRequiredPipelineCaching( + context, + new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(key.Split(KeySegmentSeperator)), + ct), + (predecessor, ct) => ReadSelectorsAsync(context, predecessor, ct), + predecessor => ComputeSelectorsWriteKey(_universe, fingerprint.WeakFingerprint, predecessor?.ManifestId.ValueString), + (predecessor, selectors, key, ct) => TryPublishSelectorsAsync(context, fingerprint, selectors, key, pathSetBytes, ct), + cancellationToken); + + return wfpAddded || sfpAddded + ? AddNodeResult.Added + : AddNodeResult.AlreadyExists; + } + + private async Task TryPublishSelectorsAsync( + Context context, + StrongFingerprint fingerprint, + IReadOnlyCollection selectors, + string key, + (ContentHash hash, byte[] bytes)? pathSetBytes, + CancellationToken cancellationToken) + { List pathSetTempFiles = new(); try { @@ -334,12 +364,6 @@ protected override async Task AddNodeAsync( List infos = new(); - string key = ComputeSelectorsKey(fingerprint.WeakFingerprint, forWrite: true); - - var selectors = await GetSelectors(context, fingerprint.WeakFingerprint, cancellationToken).ToHashSetAsync(cancellationToken); - - selectors.Add(fingerprint.Selector); - // TODO: limit the number of selectors we store. Dictionary extras = new(selectors.Count); @@ -364,7 +388,7 @@ protected override async Task AddNodeAsync( #pragma warning restore CA2000 #pragma warning restore IDE0079 #endif - var bytes = selector.ContentHash == pathSetBytes?.hash + byte[] bytes = selector.ContentHash == pathSetBytes?.hash ? pathSetBytes.Value.bytes : await GetBytes(context, selector.ContentHash.ToBlobIdentifier().ToDedupIdentifier(), cancellationToken); #if NETFRAMEWORK @@ -380,18 +404,19 @@ protected override async Task AddNodeAsync( PublishResult result = await WithHttpRetries( () => _manifestClient.PublishAsync(TempFolder, infos, extras, new ArtifactPublishOptions(), manifestFileOutputPath: null, cancellationToken), cacheContext: context, - message: $"Publishing content for {fingerprint}", + message: $"Publishing selectors for {fingerprint}", cancellationToken); + var cacheFingerprint = new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(key.Split(KeySegmentSeperator)); CreatePipelineCacheArtifactContract entry = new( DomainId, - new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(key.Split(KeySegmentSeperator)), + cacheFingerprint, result.ManifestId, result.RootId, result.ProofNodes, ContentFormatConstants.Files); - wfpAddded = await WithHttpRetries( + return await WithHttpRetries( async () => { try @@ -406,16 +431,12 @@ protected override async Task AddNodeAsync( } }, cacheContext: context, - message: $"Storing cache key for {fingerprint}", + message: $"Storing selector cache key for {fingerprint}", cancellationToken); - - return wfpAddded || sfpAddded - ? AddNodeResult.Added - : AddNodeResult.AlreadyExists; } finally { - foreach (var pathSetTempFile in pathSetTempFiles) + foreach (TempFile pathSetTempFile in pathSetTempFiles) { pathSetTempFile.Dispose(); } @@ -618,24 +639,37 @@ protected override async IAsyncEnumerable GetSelectors( { await _startupTask; - string key = ComputeSelectorsKey(fingerprint, forWrite: false); PipelineCacheArtifact? result = await QueryPipelineCaching( context, - new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(key.Split(KeySegmentSeperator)), + new VisualStudio.Services.PipelineCache.WebApi.Fingerprint(ComputeSelectorsReadKey(_universe, fingerprint).Split(KeySegmentSeperator)), cancellationToken); + foreach (Selector selector in await ReadSelectorsAsync(context, result, cancellationToken)) + { + yield return selector; + } + } + + private async Task> ReadSelectorsAsync( + Context context, + PipelineCacheArtifact? result, + CancellationToken cancellationToken) + { if (result == null) { - yield break; + return []; } + HashSet selectors = new(); using var manifestStream = new MemoryStream(await GetBytes(context, result.ManifestId, cancellationToken)); Manifest manifest = JsonSerializer.Deserialize(manifestStream)!; foreach (ManifestItem selectorItem in manifest.Items.Where(i => i.Path.StartsWith(SelectorsRelativePathBase, StringComparison.Ordinal))) { string[] tokens = selectorItem.Path.Substring(SelectorsRelativePathBase.Length + 1).Split('/'); - yield return new Selector(new ContentHash(tokens[0]), HexUtilities.HexToBytes(tokens[1])); + selectors.Add(new Selector(new ContentHash(tokens[0]), HexUtilities.HexToBytes(tokens[1]))); } + + return selectors; } protected override async Task OpenStreamAsync(Context context, ContentHash contentHash, CancellationToken cancellationToken) @@ -710,14 +744,34 @@ string message } private string ComputeKey(StrongFingerprint sfp, bool forWrite) => - forWrite - ? $"outputs{InternalSeed}{KeySegmentSeperator}{_universe}{KeySegmentSeperator}{sfp.WeakFingerprint.Serialize()}{KeySegmentSeperator}{sfp.Selector.ContentHash.Serialize()}{KeySegmentSeperator}{DateTime.UtcNow.Ticks}" - : $"outputs{InternalSeed}{KeySegmentSeperator}{_universe}{KeySegmentSeperator}{sfp.WeakFingerprint.Serialize()}{KeySegmentSeperator}{sfp.Selector.ContentHash.Serialize()}{KeySegmentSeperator}**"; + ComputeOutputKey(_universe, sfp, forWrite, DateTime.UtcNow.Ticks); - private string ComputeSelectorsKey(BuildXL.Cache.MemoizationStore.Interfaces.Sessions.Fingerprint wfp, bool forWrite) => + internal static string ComputeOutputKey(string universe, StrongFingerprint sfp, bool forWrite, long writeId) => forWrite - ? $"selector{InternalSeed}{KeySegmentSeperator}{_universe}{KeySegmentSeperator}{wfp.Serialize()}{KeySegmentSeperator}{DateTime.UtcNow.Ticks}" - : $"selector{InternalSeed}{KeySegmentSeperator}{_universe}{KeySegmentSeperator}{wfp.Serialize()}{KeySegmentSeperator}**"; + ? $"outputs{InternalSeed}{KeySegmentSeperator}{universe}{KeySegmentSeperator}{sfp.WeakFingerprint.Serialize()}{KeySegmentSeperator}{sfp.Selector.ContentHash.Serialize()}{KeySegmentSeperator}{sfp.Selector.Output.ToHexString()}{KeySegmentSeperator}{writeId}" + : $"outputs{InternalSeed}{KeySegmentSeperator}{universe}{KeySegmentSeperator}{sfp.WeakFingerprint.Serialize()}{KeySegmentSeperator}{sfp.Selector.ContentHash.Serialize()}{KeySegmentSeperator}{sfp.Selector.Output.ToHexString()}{KeySegmentSeperator}**"; + + internal static string ComputeSelectorsReadKey(string universe, BuildXL.Cache.MemoizationStore.Interfaces.Sessions.Fingerprint wfp) => + $"selector{InternalSeed}{KeySegmentSeperator}{universe}{KeySegmentSeperator}{wfp.Serialize()}{KeySegmentSeperator}**"; + + internal static string ComputeSelectorsWriteKey( + string universe, + BuildXL.Cache.MemoizationStore.Interfaces.Sessions.Fingerprint wfp, + string? predecessorManifestId) => + $"selector{InternalSeed}{KeySegmentSeperator}{universe}{KeySegmentSeperator}{wfp.Serialize()}{KeySegmentSeperator}{predecessorManifestId ?? "root"}"; + + private Task QueryRequiredPipelineCaching( + Context context, + VisualStudio.Services.PipelineCache.WebApi.Fingerprint key, + CancellationToken cancellationToken) + { + return WithHttpRetries( + async () => await QueryPipelineCaching(context, key, cancellationToken) + ?? throw new CacheException($"Pipeline Cache entry `{key}` was reported as existing but is not yet visible."), + cacheContext: context, + message: $"Querying required cache entry '{key}'", + cancellationToken); + } private Task QueryPipelineCaching(Context context, VisualStudio.Services.PipelineCache.WebApi.Fingerprint key, CancellationToken cancellationToken) { diff --git a/src/AzurePipelines/SelectorPublication.cs b/src/AzurePipelines/SelectorPublication.cs new file mode 100644 index 0000000..e428ecd --- /dev/null +++ b/src/AzurePipelines/SelectorPublication.cs @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using BuildXL.Cache.MemoizationStore.Interfaces.Sessions; + +namespace Microsoft.MSBuildCache.AzurePipelines; + +internal static class SelectorPublication +{ + internal static async Task AddAsync( + Selector selector, + string latestKey, + Func> query, + Func> getConflictWinner, + Func>> getSelectors, + Func getWriteKey, + Func, string, CancellationToken, Task> tryPublish, + CancellationToken cancellationToken) + where TState : class + { + TState? predecessor = await query(latestKey, cancellationToken); + + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + HashSet selectors = new(await getSelectors(predecessor, cancellationToken)); + if (!selectors.Add(selector)) + { + return false; + } + + string writeKey = getWriteKey(predecessor); + if (await tryPublish(predecessor, selectors, writeKey, cancellationToken)) + { + return true; + } + + predecessor = await getConflictWinner(writeKey, cancellationToken); + } + } +} diff --git a/src/Common.Tests/CasCacheClientTests.cs b/src/Common.Tests/CasCacheClientTests.cs new file mode 100644 index 0000000..9a4fff5 --- /dev/null +++ b/src/Common.Tests/CasCacheClientTests.cs @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +using System; +using BuildXL.Cache.ContentStore.Hashing; +using BuildXL.Cache.MemoizationStore.Interfaces.Results; +using BuildXL.Cache.MemoizationStore.Interfaces.Sessions; +using Microsoft.MSBuildCache.Caching; +using Microsoft.VisualStudio.TestTools.UnitTesting; + +namespace Microsoft.MSBuildCache.Tests; + +[TestClass] +public class CasCacheClientTests +{ + [TestMethod] + public void NullContentHashListMeansSubmittedValueWasAccepted() + { + AddOrGetContentHashListResult result = new(default(ContentHashListWithDeterminism)); + + Assert.AreEqual(AddNodeResult.Added, CasCacheClient.GetAddNodeResult(result)); + } + + [TestMethod] + public void ReturnedContentHashListMeansAnotherValueWon() + { + ContentHashList contentHashList = new(Array.Empty(), null); + AddOrGetContentHashListResult result = new(new ContentHashListWithDeterminism(contentHashList, CacheDeterminism.None)); + + Assert.AreEqual(AddNodeResult.AlreadyExists, CasCacheClient.GetAddNodeResult(result)); + } +} diff --git a/src/Common/Caching/CasCacheClient.cs b/src/Common/Caching/CasCacheClient.cs index f9ca237..f52f6de 100644 --- a/src/Common/Caching/CasCacheClient.cs +++ b/src/Common/Caching/CasCacheClient.cs @@ -253,13 +253,18 @@ static async Task checkUploadResultsAsync(List> uploadTas throw new CacheException($"{nameof(_twoLevelCacheSession.AddOrGetContentHashListAsync)} failed for {fingerprint}."); } - return contentHashList.Equals(addResult?.ContentHashListWithDeterminism.ContentHashList) - ? AddNodeResult.Added - : AddNodeResult.AlreadyExists; + return GetAddNodeResult(addResult); // TODO dfederm: Handle CHL races } + internal static AddNodeResult GetAddNodeResult(AddOrGetContentHashListResult addResult) + { + return addResult.ContentHashListWithDeterminism.ContentHashList == null + ? AddNodeResult.Added + : AddNodeResult.AlreadyExists; + } + protected override async Task GetCacheEntryAsync( Context context, StrongFingerprint cacheStrongFingerprint,