<commit_before>/*
 * Copyright 2017-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License. You may obtain
 * a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package com.facebook.buck.artifact_cache;

import com.facebook.buck.io.file.BorrowablePath;
import com.facebook.buck.io.file.LazyPath;
import com.facebook.buck.io.filesystem.ProjectFilesystem;
import com.facebook.buck.log.Logger;
import com.facebook.buck.rules.RuleKey;
import com.facebook.buck.util.MoreCollectors;
import com.facebook.buck.util.Scope;
import com.facebook.buck.util.immutables.BuckStyleTuple;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
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.ListeningExecutorService;
import com.google.common.util.concurrent.SettableFuture;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
import org.immutables.value.Value;

public abstract class AbstractAsynchronousCache implements ArtifactCache {
  private static final Logger LOG = Logger.get(AbstractAsynchronousCache.class);
  private static final int MAX_CONSECUTIVE_MULTI_FETCH_ERRORS = 3;
  private final String name;
  private final CacheReadMode cacheReadMode;

  private final ListeningExecutorService storeExecutorService;
  // TODO(cjhopman): The fetch handling would probably be simpler if we created a threadpool
  // ourselves and just ran a bunch of threads just continuously running processFetch() and have
  // that just take() FetchRequests off a BlockingQueue.
  private final ListeningExecutorService fetchExecutorService;
  private final CacheEventListener eventListener;

  private final Optional<Long> maxStoreSize;
  private final ProjectFilesystem projectFilesystem;
  private final ArtifactCacheMode mode;

  private final BlockingQueue<FetchRequest> pendingFetchRequests = new LinkedBlockingQueue<>();

  // TODO(cjhopman): Remove this error-based disabling of multiFetch, it's only here to make rollout
  // less disruptive.
  private volatile boolean enableMultiFetch = true;
  private final AtomicInteger consecutiveMultiFetchErrorCount = new AtomicInteger();

  public AbstractAsynchronousCache(
      String name,
      ArtifactCacheMode mode,
      CacheReadMode cacheReadMode,
      ListeningExecutorService storeExecutorService,
      ListeningExecutorService fetchExecutorService,
      CacheEventListener eventListener,
      Optional<Long> maxStoreSize,
      ProjectFilesystem projectFilesystem) {
    this.name = name;
    this.cacheReadMode = cacheReadMode;
    this.storeExecutorService = storeExecutorService;
    this.fetchExecutorService = fetchExecutorService;
    this.eventListener = eventListener;
    this.maxStoreSize = maxStoreSize;
    this.projectFilesystem = projectFilesystem;
    this.mode = mode;
  }

  protected final String getName() {
    return name;
  }

  protected final ArtifactCacheMode getMode() {
    return mode;
  }

  protected final ProjectFilesystem getProjectFilesystem() {
    return projectFilesystem;
  }

  protected abstract FetchResult fetchImpl(RuleKey ruleKey, LazyPath output) throws IOException;

  protected abstract StoreResult storeImpl(ArtifactInfo info, Path file) throws IOException;

  /** The MultiFetchResult should contain results in the same order as the requests. */
  protected abstract MultiFetchResult multiFetchImpl(Iterable<FetchRequest> requests)
      throws IOException;

  /**
   * Used to compute the number of keys to include in every multiFetchRequest. If < 1, fetch will be
   * used instead of multifetch.
   */
  @SuppressWarnings("unused")
  protected int getMultiFetchBatchSize(int pendingRequestsSize) {
    return 0;
  }

  private void doMultiFetch(ImmutableList<ClaimedFetchRequest> requests) {
    boolean gotNonError = false;
    try (CacheEventListener.MultiFetchRequestEvents requestEvents =
        eventListener.multiFetchStarted(
            requests
                .stream()
                .map(r -> r.getRequest().getRuleKey())
                .collect(MoreCollectors.toImmutableList()))) {
      try {
        MultiFetchResult result =
            multiFetchImpl(
                requests
                    .stream()
                    .map(ClaimedFetchRequest::getRequest)
                    .collect(MoreCollectors.toImmutableList()));
        Preconditions.checkState(result.getResults().size() == requests.size());
        // MultiFetch must return a non-skipped result for at least one of the requested keys.
        Preconditions.checkState(
            result
                .getResults()
                .stream()
                .anyMatch(
                    fetchResult ->
                        fetchResult.getCacheResult().getType() != CacheResultType.SKIPPED));
        for (int i = 0; i < requests.size(); i++) {
          ClaimedFetchRequest thisRequest = requests.get(i);
          FetchResult thisResult = result.getResults().get(i);
          if (thisResult.getCacheResult().getType() == CacheResultType.SKIPPED) {
            requestEvents.skipped(i);
            thisRequest.reschedule();
          } else {
            requestEvents.finished(i, thisResult);
            thisRequest.setResult(thisResult.getCacheResult());
          }
        }
        gotNonError =
            result
                .getResults()
                .stream()
                .anyMatch(
                    fetchResult -> fetchResult.getCacheResult().getType() != CacheResultType.ERROR);
      } catch (IOException e) {
        ImmutableList<RuleKey> keys =
            requests
                .stream()
                .map(r -> r.getRequest().getRuleKey())
                .collect(MoreCollectors.toImmutableList());
        String msg =
            String.format(
                "multifetch(<%s>): %s: %s",
                Joiner.on(", ").join(keys), e.getClass().getName(), e.getMessage());
        // Some of these might already be fulfilled. That's fine, this set() call will just be
        // ignored.
        for (int i = 0; i < requests.size(); i++) {
          CacheResult result = CacheResult.error(name, mode, msg);
          requestEvents.failed(i, e, msg, result);
          requests.get(i).setResult(result);
        }
      }
    } finally {
      if (gotNonError) {
        consecutiveMultiFetchErrorCount.set(0);
      } else {
        if (consecutiveMultiFetchErrorCount.incrementAndGet()
            == MAX_CONSECUTIVE_MULTI_FETCH_ERRORS) {
          LOG.error("Too many MultiFetch errors, falling back to Fetch only.");
          enableMultiFetch = false;
        }
      }
    }
  }

  private void doFetch(FetchRequest request) {
    CacheResult result;
    CacheEventListener.FetchRequestEvents requestEvents =
        eventListener.fetchStarted(request.getRuleKey());
    try {
      FetchResult fetchResult = fetchImpl(request.getRuleKey(), request.getOutput());
      result = fetchResult.getCacheResult();
      requestEvents.finished(fetchResult);
    } catch (IOException e) {
      String msg =
          String.format(
              "fetch(%s): %s: %s", request.getRuleKey(), e.getClass().getName(), e.getMessage());
      result = CacheResult.error(name, mode, msg);<commit_msg>Demote multifetch error to info<commit_after>/*
 * Copyright 2017-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License. You may obtain
 * a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package com.facebook.buck.artifact_cache;

import com.facebook.buck.io.file.BorrowablePath;
import com.facebook.buck.io.file.LazyPath;
import com.facebook.buck.io.filesystem.ProjectFilesystem;
import com.facebook.buck.log.Logger;
import com.facebook.buck.rules.RuleKey;
import com.facebook.buck.util.MoreCollectors;
import com.facebook.buck.util.Scope;
import com.facebook.buck.util.immutables.BuckStyleTuple;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
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.ListeningExecutorService;
import com.google.common.util.concurrent.SettableFuture;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
import org.immutables.value.Value;

public abstract class AbstractAsynchronousCache implements ArtifactCache {
  private static final Logger LOG = Logger.get(AbstractAsynchronousCache.class);
  private static final int MAX_CONSECUTIVE_MULTI_FETCH_ERRORS = 3;
  private final String name;
  private final CacheReadMode cacheReadMode;

  private final ListeningExecutorService storeExecutorService;
  // TODO(cjhopman): The fetch handling would probably be simpler if we created a threadpool
  // ourselves and just ran a bunch of threads just continuously running processFetch() and have
  // that just take() FetchRequests off a BlockingQueue.
  private final ListeningExecutorService fetchExecutorService;
  private final CacheEventListener eventListener;

  private final Optional<Long> maxStoreSize;
  private final ProjectFilesystem projectFilesystem;
  private final ArtifactCacheMode mode;

  private final BlockingQueue<FetchRequest> pendingFetchRequests = new LinkedBlockingQueue<>();

  // TODO(cjhopman): Remove this error-based disabling of multiFetch, it's only here to make rollout
  // less disruptive.
  private volatile boolean enableMultiFetch = true;
  private final AtomicInteger consecutiveMultiFetchErrorCount = new AtomicInteger();

  public AbstractAsynchronousCache(
      String name,
      ArtifactCacheMode mode,
      CacheReadMode cacheReadMode,
      ListeningExecutorService storeExecutorService,
      ListeningExecutorService fetchExecutorService,
      CacheEventListener eventListener,
      Optional<Long> maxStoreSize,
      ProjectFilesystem projectFilesystem) {
    this.name = name;
    this.cacheReadMode = cacheReadMode;
    this.storeExecutorService = storeExecutorService;
    this.fetchExecutorService = fetchExecutorService;
    this.eventListener = eventListener;
    this.maxStoreSize = maxStoreSize;
    this.projectFilesystem = projectFilesystem;
    this.mode = mode;
  }

  protected final String getName() {
    return name;
  }

  protected final ArtifactCacheMode getMode() {
    return mode;
  }

  protected final ProjectFilesystem getProjectFilesystem() {
    return projectFilesystem;
  }

  protected abstract FetchResult fetchImpl(RuleKey ruleKey, LazyPath output) throws IOException;

  protected abstract StoreResult storeImpl(ArtifactInfo info, Path file) throws IOException;

  /** The MultiFetchResult should contain results in the same order as the requests. */
  protected abstract MultiFetchResult multiFetchImpl(Iterable<FetchRequest> requests)
      throws IOException;

  /**
   * Used to compute the number of keys to include in every multiFetchRequest. If < 1, fetch will be
   * used instead of multifetch.
   */
  @SuppressWarnings("unused")
  protected int getMultiFetchBatchSize(int pendingRequestsSize) {
    return 0;
  }

  private void doMultiFetch(ImmutableList<ClaimedFetchRequest> requests) {
    boolean gotNonError = false;
    try (CacheEventListener.MultiFetchRequestEvents requestEvents =
        eventListener.multiFetchStarted(
            requests
                .stream()
                .map(r -> r.getRequest().getRuleKey())
                .collect(MoreCollectors.toImmutableList()))) {
      try {
        MultiFetchResult result =
            multiFetchImpl(
                requests
                    .stream()
                    .map(ClaimedFetchRequest::getRequest)
                    .collect(MoreCollectors.toImmutableList()));
        Preconditions.checkState(result.getResults().size() == requests.size());
        // MultiFetch must return a non-skipped result for at least one of the requested keys.
        Preconditions.checkState(
            result
                .getResults()
                .stream()
                .anyMatch(
                    fetchResult ->
                        fetchResult.getCacheResult().getType() != CacheResultType.SKIPPED));
        for (int i = 0; i < requests.size(); i++) {
          ClaimedFetchRequest thisRequest = requests.get(i);
          FetchResult thisResult = result.getResults().get(i);
          if (thisResult.getCacheResult().getType() == CacheResultType.SKIPPED) {
            requestEvents.skipped(i);
            thisRequest.reschedule();
          } else {
            requestEvents.finished(i, thisResult);
            thisRequest.setResult(thisResult.getCacheResult());
          }
        }
        gotNonError =
            result
                .getResults()
                .stream()
                .anyMatch(
                    fetchResult -> fetchResult.getCacheResult().getType() != CacheResultType.ERROR);
      } catch (IOException e) {
        ImmutableList<RuleKey> keys =
            requests
                .stream()
                .map(r -> r.getRequest().getRuleKey())
                .collect(MoreCollectors.toImmutableList());
        String msg =
            String.format(
                "multifetch(<%s>): %s: %s",
                Joiner.on(", ").join(keys), e.getClass().getName(), e.getMessage());
        // Some of these might already be fulfilled. That's fine, this set() call will just be
        // ignored.
        for (int i = 0; i < requests.size(); i++) {
          CacheResult result = CacheResult.error(name, mode, msg);
          requestEvents.failed(i, e, msg, result);
          requests.get(i).setResult(result);
        }
      }
    } finally {
      if (gotNonError) {
        consecutiveMultiFetchErrorCount.set(0);
      } else {
        if (consecutiveMultiFetchErrorCount.incrementAndGet()
            == MAX_CONSECUTIVE_MULTI_FETCH_ERRORS) {
          LOG.info("Too many MultiFetch errors, falling back to Fetch only.");
          enableMultiFetch = false;
        }
      }
    }
  }

  private void doFetch(FetchRequest request) {
    CacheResult result;
    CacheEventListener.FetchRequestEvents requestEvents =
        eventListener.fetchStarted(request.getRuleKey());
    try {
      FetchResult fetchResult = fetchImpl(request.getRuleKey(), request.getOutput());
      result = fetchResult.getCacheResult();
      requestEvents.finished(fetchResult);
    } catch (IOException e) {
      String msg =
          String.format(
              "fetch(%s): %s: %s", request.getRuleKey(), e.getClass().getName(), e.getMessage());
      result = CacheResult.error(name, mode, msg);
{'fraction_non_alphanumeric': 0.08919403723634564, 'fraction_numerical': 0.0017319230531329252, 'mean_word_length': 3.0838595604950747, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5527874', 'n_tokens_mistral': 4256, 'n_tokens_neox': 4140, 'n_words': 1105}