Fix possible race condition when saving binary to cache

Summary: When we uploading artifacts to buck, there is a possibility that the metadata will be uploaded before content. This might cause another worker to read metadata, fail to load content, and reupload the artifact (metadata + content). With high volume of reads a writes we can end up reloading the same artifact over and over again.

Reviewed By: ttsugriy

fbshipit-source-id: b3f06b7

src/com/facebook/buck/artifact_cache/AbstractAsynchronousCache.java

@@ -26,6 +26,7 @@
 import com.facebook.buck.io.filesystem.ProjectFilesystem;
 import com.facebook.buck.log.Logger;
 import com.facebook.buck.util.Scope;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
@@ -402,6 +403,84 @@ public final ListenableFuture<Void> store(ArtifactInfo info, BorrowablePath outp
         });
   }
 
+  @Override
+  public final ListenableFuture<Void> store(
+      ImmutableList<Pair<ArtifactInfo, BorrowablePath>> artifacts) {
+    if (!getCacheReadMode().isWritable()) {
+      return Futures.immediateFuture(null);
+    }
+
+    ImmutableList.Builder<Pair<ArtifactInfo, Path>> matchedArtifactsBuilder =
+        ImmutableList.builderWithExpectedSize(artifacts.size());
+    ImmutableList.Builder<Long> artifactSizesInBytesBuilder =
+        ImmutableList.builderWithExpectedSize(artifacts.size());
+
+    for (int i = 0; i < artifacts.size(); i++) {
+      BorrowablePath output = artifacts.get(i).getSecond();
+      ArtifactInfo info = artifacts.get(i).getFirst();
+      long artifactSizeBytes = getFileSize(output.getPath());
+      if (artifactExceedsMaximumSize(artifactSizeBytes)) {
+        LOG.info(
+            "Artifact too big so not storing it in the %s cache. file=[%s] buildTarget=[%s]",
+            name, output.getPath(), info.getBuildTarget());
+        continue;
+      }
+
+      Path tmp;
+      try {
+        tmp = getPathForArtifact(output);
+      } catch (IOException e) {
+        LOG.error(e, "Failed to store artifact in temp file: " + output.getPath());
+        continue;
+      }
+
+      matchedArtifactsBuilder.add(new Pair<>(info, tmp));
+      artifactSizesInBytesBuilder.add(artifactSizeBytes);
+    }
+
+    ImmutableList<Pair<ArtifactInfo, Path>> matchedArtifacts = matchedArtifactsBuilder.build();
+
+    if (matchedArtifacts.isEmpty()) {
+      return Futures.immediateFuture(null);
+    }
+
+    ImmutableList<Long> artifactSizesInBytes = artifactSizesInBytesBuilder.build();
+
+    ImmutableList.Builder<StoreEvents> eventsBuilder =
+        ImmutableList.builderWithExpectedSize(artifactSizesInBytes.size());
+    for (int i = 0; i < artifactSizesInBytes.size(); i++) {
+      eventsBuilder.add(
+          eventListener.storeScheduled(
+              matchedArtifacts.get(i).getFirst(), artifactSizesInBytes.get(i)));
+    }
+
+    ImmutableList<StoreEvents> events = eventsBuilder.build();
+
+    return storeExecutorService.submit(
+        () -> {
+          for (int i = 0; i < matchedArtifacts.size(); i++) {
+            StoreEvents.StoreRequestEvents requestEvents = events.get(i).started();
+            try {
+              StoreResult result =
+                  storeImpl(
+                      matchedArtifacts.get(i).getFirst(), matchedArtifacts.get(i).getSecond());
+              requestEvents.finished(result);
+            } catch (IOException e) {
+              String msg =
+                  String.format(
+                      "store(%s): %s: %s",
+                      matchedArtifacts.get(i).getFirst().getRuleKeys(),
+                      e.getClass().getName(),
+                      e.getMessage());
+              requestEvents.failed(e, msg);
+              throw new RuntimeException(e);
+            }
+          }
+
+          return null;
+        });
+  }
+
   @Override
   public final ListenableFuture<CacheDeleteResult> deleteAsync(List<RuleKey> ruleKeys) {
     if (!getCacheReadMode().isWritable()) {

src/com/facebook/buck/artifact_cache/ArtifactCache.java

@@ -21,9 +21,13 @@
 import com.facebook.buck.core.rulekey.RuleKey;
 import com.facebook.buck.io.file.BorrowablePath;
 import com.facebook.buck.io.file.LazyPath;
+import com.facebook.buck.util.types.Pair;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import java.util.List;
 import javax.annotation.Nullable;
 
@@ -59,6 +63,31 @@ ListenableFuture<CacheResult> fetchAsync(
    */
   ListenableFuture<Void> store(ArtifactInfo info, BorrowablePath output);
 
+  /**
+   * Store the list of artifacts at path specified by output to cache in passed order, such that it
+   * can later be fetched using ruleKey as the lookup key. If any internal errors occur, fail
+   * silently and continue execution. Store may be performed synchronously or asynchronously.
+   *
+   * <p>This is a noop if {@link #getCacheReadMode()}} returns {@code READONLY}.
+   *
+   * @param artifacts list of artifact info and path to be uploaded to the cache in given order.
+   * @return {@link ListenableFuture} that completes once the store has finished.
+   */
+  default ListenableFuture<Void> store(
+      ImmutableList<Pair<ArtifactInfo, BorrowablePath>> artifacts) {
+    if (artifacts.isEmpty()) {
+      return Futures.immediateFuture(null);
+    }
+
+    Pair<ArtifactInfo, BorrowablePath> first = artifacts.get(0);
+    ImmutableList<Pair<ArtifactInfo, BorrowablePath>> rest = artifacts.subList(1, artifacts.size());
+
+    return Futures.transformAsync(
+        this.store(first.getFirst(), first.getSecond()),
+        input -> this.store(rest),
+        MoreExecutors.directExecutor());
+  }
+
   /**
    * Check if the cache contains the given artifacts, keyed by ruleKeys, without fetching them, and
    * return a map of results wrapped in a {@link ListenableFuture}. This is supposed to be fast, but

src/com/facebook/buck/artifact_cache/MultiArtifactCache.java

@@ -21,6 +21,7 @@
 import com.facebook.buck.core.rulekey.RuleKey;
 import com.facebook.buck.io.file.BorrowablePath;
 import com.facebook.buck.io.file.LazyPath;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Functions;
 import com.google.common.collect.ImmutableList;
@@ -130,6 +131,30 @@ public ListenableFuture<Void> store(ArtifactInfo info, BorrowablePath output) {
     return storeToCaches(writableArtifactCaches, info, output);
   }
 
+  @Override
+  public ListenableFuture<Void> store(ImmutableList<Pair<ArtifactInfo, BorrowablePath>> artifacts) {
+    if (writableArtifactCaches.size() != 1) {
+      ImmutableList.Builder<Pair<ArtifactInfo, BorrowablePath>> artifactTemporaryPaths =
+          ImmutableList.builderWithExpectedSize(artifacts.size());
+      for (int i = 0; i < artifacts.size(); i++) {
+        artifactTemporaryPaths.add(
+            new Pair<>(
+                artifacts.get(i).getFirst(),
+                BorrowablePath.notBorrowablePath(artifacts.get(i).getSecond().getPath())));
+      }
+      artifacts = artifactTemporaryPaths.build();
+    }
+
+    List<ListenableFuture<Void>> storeFutures =
+        Lists.newArrayListWithExpectedSize(writableArtifactCaches.size());
+    for (ArtifactCache artifactCache : writableArtifactCaches) {
+      storeFutures.add(artifactCache.store(artifacts));
+    }
+
+    // Aggregate future to ensure all store operations have completed.
+    return Futures.transform(Futures.allAsList(storeFutures), Functions.constant(null));
+  }
+
   @Override
   public ListenableFuture<ImmutableMap<RuleKey, CacheResult>> multiContainsAsync(
       ImmutableSet<RuleKey> ruleKeys) {

src/com/facebook/buck/artifact_cache/NoopArtifactCache.java

@@ -22,6 +22,7 @@
 import com.facebook.buck.core.rulekey.RuleKey;
 import com.facebook.buck.io.file.BorrowablePath;
 import com.facebook.buck.io.file.LazyPath;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -49,6 +50,11 @@ public ListenableFuture<Void> store(ArtifactInfo info, BorrowablePath output) {
     return Futures.immediateFuture(null);
   }
 
+  @Override
+  public ListenableFuture<Void> store(ImmutableList<Pair<ArtifactInfo, BorrowablePath>> artifacts) {
+    return Futures.immediateFuture(null);
+  }
+
   @Override
   public ListenableFuture<ImmutableMap<RuleKey, CacheResult>> multiContainsAsync(
       ImmutableSet<RuleKey> ruleKeys) {

src/com/facebook/buck/artifact_cache/RetryingCacheDecorator.java

@@ -26,7 +26,9 @@
 import com.facebook.buck.io.file.LazyPath;
 import com.facebook.buck.log.Logger;
 import com.facebook.buck.slb.NoHealthyServersException;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.Futures;
@@ -110,6 +112,11 @@ public ListenableFuture<Void> store(ArtifactInfo info, BorrowablePath output) {
     return delegate.store(info, output);
   }
 
+  @Override
+  public ListenableFuture<Void> store(ImmutableList<Pair<ArtifactInfo, BorrowablePath>> artifacts) {
+    return delegate.store(artifacts);
+  }
+
   @Override
   public ListenableFuture<ImmutableMap<RuleKey, CacheResult>> multiContainsAsync(
       ImmutableSet<RuleKey> ruleKeys) {

src/com/facebook/buck/artifact_cache/TwoLevelArtifactCacheDecorator.java

@@ -30,6 +30,7 @@
 import com.facebook.buck.io.filesystem.ProjectFilesystem;
 import com.facebook.buck.log.Logger;
 import com.facebook.buck.util.RichStream;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Functions;
 import com.google.common.collect.ImmutableList;
@@ -249,17 +250,27 @@ private ListenableFuture<Boolean> attemptTwoLevelStore(ArtifactInfo info, Borrow
             .putAll(info.getMetadata())
             .put(METADATA_KEY, hashCode)
             .build();
+    // We need to upload artifacts in this order to prevent race condition. If we would do
+    // it concurrently it is possible that we upload metadata before the file. Then other
+    // builder read metadata, but cannot find a file (which is still being uploaded), and
+    // decide that we have to re-upload it. With enough machines building the same target
+    // we end up with constant re-uploading and rebuilding flow. The following issue is
+    // only in case when output hash changes between builds.
+    Pair<ArtifactInfo, BorrowablePath> artifact =
+        new Pair<>(ArtifactInfo.builder().addRuleKeys(new RuleKey(hashCode)).build(), output);
+    Pair<ArtifactInfo, BorrowablePath> metadata =
+        new Pair<>(
+            ArtifactInfo.builder()
+                .setRuleKeys(info.getRuleKeys())
+                .setMetadata(metadataWithCacheKey)
+                .build(),
+            BorrowablePath.notBorrowablePath(emptyFilePath));
 
     return Futures.transform(
-        Futures.allAsList(
-            delegate.store(
-                ArtifactInfo.builder()
-                    .setRuleKeys(info.getRuleKeys())
-                    .setMetadata(metadataWithCacheKey)
-                    .build(),
-                BorrowablePath.notBorrowablePath(emptyFilePath)),
-            delegate.store(
-                ArtifactInfo.builder().addRuleKeys(new RuleKey(hashCode)).build(), output)),
+        // This relies on the fact that delegate stores artifacts in sequential way in the order
+        // they are being passed. If we store them internally in consecutive way, there is a
+        // possibility of race condition.
+        delegate.store(ImmutableList.of(artifact, metadata)),
         Functions.constant(true),
         MoreExecutors.directExecutor());
   }

test/com/facebook/buck/artifact_cache/DirArtifactCacheTest.java

@@ -47,6 +47,7 @@
 import com.facebook.buck.testutil.TemporaryPaths;
 import com.facebook.buck.util.DirectoryCleaner;
 import com.facebook.buck.util.cache.FileHashCache;
+import com.facebook.buck.util.types.Pair;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -386,6 +387,119 @@ public void testCacheStoresAndFetchHits() throws IOException {
     }
   }
 
+  @Test
+  public void testCacheMultiStoresAndContainsHits() throws IOException {
+    Path cacheDir = tmpDir.newFolder();
+    Path fileX = tmpDir.newFile("x");
+    Path fileY = tmpDir.newFile("y");
+    Path fileZ = tmpDir.newFile("z");
+
+    fileHashCache =
+        new FakeFileHashCache(
+            ImmutableMap.of(
+                fileX, HashCode.fromInt(0),
+                fileY, HashCode.fromInt(1),
+                fileZ, HashCode.fromInt(2)));
+
+    dirArtifactCache =
+        new DirArtifactCache(
+            "dir",
+            TestProjectFilesystems.createProjectFilesystem(cacheDir),
+            Paths.get("."),
+            CacheReadMode.READWRITE,
+            /* maxCacheSizeBytes */ Optional.empty());
+
+    Files.write(fileX, "x".getBytes(UTF_8));
+    Files.write(fileY, "y".getBytes(UTF_8));
+    Files.write(fileZ, "x".getBytes(UTF_8));
+
+    BuildRule inputRuleX = new BuildRuleForTest(fileX);
+    BuildRule inputRuleY = new BuildRuleForTest(fileY);
+    BuildRule inputRuleZ = new BuildRuleForTest(fileZ);
+    assertFalse(inputRuleX.equals(inputRuleY));
+    assertFalse(inputRuleX.equals(inputRuleZ));
+    assertFalse(inputRuleY.equals(inputRuleZ));
+    ActionGraphBuilder graphBuilder = new TestActionGraphBuilder();
+    graphBuilder.addToIndex(inputRuleX);
+    graphBuilder.addToIndex(inputRuleY);
+    graphBuilder.addToIndex(inputRuleZ);
+    SourcePathRuleFinder ruleFinder = new SourcePathRuleFinder(graphBuilder);
+    SourcePathResolver resolver = DefaultSourcePathResolver.from(ruleFinder);
+
+    DefaultRuleKeyFactory fakeRuleKeyFactory =
+        new TestDefaultRuleKeyFactory(fileHashCache, resolver, ruleFinder);
+
+    RuleKey ruleKeyX = fakeRuleKeyFactory.build(inputRuleX);
+    RuleKey ruleKeyY = fakeRuleKeyFactory.build(inputRuleY);
+    RuleKey ruleKeyZ = fakeRuleKeyFactory.build(inputRuleZ);
+
+    assertEquals(
+        CacheResultType.MISS,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyX, LazyPath.ofInstance(fileX)))
+            .getType());
+    assertEquals(
+        CacheResultType.MISS,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyY, LazyPath.ofInstance(fileY)))
+            .getType());
+    assertEquals(
+        CacheResultType.MISS,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyZ, LazyPath.ofInstance(fileZ)))
+            .getType());
+
+    dirArtifactCache.store(
+        ImmutableList.of(
+            new Pair<>(
+                ArtifactInfo.builder().addRuleKeys(ruleKeyX).build(),
+                BorrowablePath.notBorrowablePath(fileX)),
+            new Pair<>(
+                ArtifactInfo.builder().addRuleKeys(ruleKeyY).build(),
+                BorrowablePath.notBorrowablePath(fileY)),
+            new Pair<>(
+                ArtifactInfo.builder().addRuleKeys(ruleKeyZ).build(),
+                BorrowablePath.notBorrowablePath(fileZ))));
+
+    Files.delete(fileX);
+    Files.delete(fileY);
+    Files.delete(fileZ);
+
+    assertEquals(
+        CacheResultType.HIT,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyX, LazyPath.ofInstance(fileX)))
+            .getType());
+    assertEquals(
+        CacheResultType.HIT,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyY, LazyPath.ofInstance(fileY)))
+            .getType());
+    assertEquals(
+        CacheResultType.HIT,
+        Futures.getUnchecked(
+                dirArtifactCache.fetchAsync(null, ruleKeyZ, LazyPath.ofInstance(fileZ)))
+            .getType());
+
+    assertEquals(inputRuleX, new BuildRuleForTest(fileX));
+    assertEquals(inputRuleY, new BuildRuleForTest(fileY));
+    assertEquals(inputRuleZ, new BuildRuleForTest(fileZ));
+
+    ImmutableList<Path> cachedFiles = ImmutableList.copyOf(dirArtifactCache.getAllFilesInCache());
+    assertEquals(6, cachedFiles.size());
+
+    ImmutableSet<String> filenames =
+        cachedFiles
+            .stream()
+            .map(input -> input.getFileName().toString())
+            .collect(ImmutableSet.toImmutableSet());
+
+    for (RuleKey ruleKey : ImmutableSet.of(ruleKeyX, ruleKeyY, ruleKeyZ)) {
+      assertThat(filenames, Matchers.hasItem(ruleKey.toString()));
+      assertThat(filenames, Matchers.hasItem(ruleKey + ".metadata"));
+    }
+  }
+
   @Test
   public void testCacheStoresAndMultiContainsHits() throws IOException {
     Path cacheDir = tmpDir.newFolder();