<commit_before>package com.hubspot.singularity.data.history;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.hubspot.mesos.JavaUtils;
import com.hubspot.singularity.SingularityDeleteResult;
import com.hubspot.singularity.SingularityPendingDeploy;
import com.hubspot.singularity.SingularityTaskHistory;
import com.hubspot.singularity.SingularityTaskId;
import com.hubspot.singularity.config.SingularityConfiguration;
import com.hubspot.singularity.data.DeployManager;
import com.hubspot.singularity.data.TaskManager;

@Singleton
public class SingularityTaskHistoryPersister extends SingularityHistoryPersister<SingularityTaskId> {

  private static final Logger LOG = LoggerFactory.getLogger(SingularityTaskHistoryPersister.class);

  private final TaskManager taskManager;
  private final DeployManager deployManager;
  private final HistoryManager historyManager;

  @Inject
  public SingularityTaskHistoryPersister(SingularityConfiguration configuration, TaskManager taskManager, DeployManager deployManager, HistoryManager historyManager) {
    super(configuration);

    this.taskManager = taskManager;
    this.historyManager = historyManager;
    this.deployManager = deployManager;
  }

  @Override
  public void runActionOnPoll() {
    LOG.info("Checking inactive task ids for task history persistence");

    final long start = System.currentTimeMillis();

    final List<SingularityTaskId> allTaskIds = taskManager.getAllTaskIds();

    final Set<SingularityTaskId> activeTaskIds = Sets.newHashSet(taskManager.getActiveTaskIds());
    final Set<SingularityTaskId> lbCleaningTaskIds = Sets.newHashSet(taskManager.getLBCleanupTasks());
    final List<SingularityPendingDeploy> pendingDeploys = deployManager.getPendingDeploys();

    int numTotal = 0;
    int numTransferred = 0;

    for (SingularityTaskId taskId : allTaskIds) {
      if (activeTaskIds.contains(taskId) || lbCleaningTaskIds.contains(taskId) || isPartofPendingDeploy(pendingDeploys, taskId)) {
        continue;
      }

      final long age = System.currentTimeMillis() - taskId.getStartedAt();

      if (age < configuration.getTaskPersistAfterStartupBufferMillis()) {
        LOG.debug("Not persisting {}, it has started up too recently {} (buffer: {}) - this prevents race conditions with ZK tx", taskId, JavaUtils.durationFromMillis(age),
            JavaUtils.durationFromMillis(configuration.getTaskPersistAfterStartupBufferMillis()));
        continue;
      }

      if (moveToHistoryOrCheckForPurge(taskId)) {
        numTransferred++;
      }

      numTotal++;
    }

    LOG.info("Transferred {} out of {} inactive task ids (total {}) in {}", numTransferred, numTotal, allTaskIds.size(), JavaUtils.duration(start));
  }

  private boolean isPartofPendingDeploy(List<SingularityPendingDeploy> pendingDeploys, SingularityTaskId taskId) {
    for (SingularityPendingDeploy pendingDeploy : pendingDeploys) {
      if (pendingDeploy.getDeployMarker().getDeployId().equals(taskId.getDeployId()) && pendingDeploy.getDeployMarker().getRequestId().equals(taskId.getRequestId())) {<commit_msg>measure vs start time<commit_after>package com.hubspot.singularity.data.history;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.hubspot.mesos.JavaUtils;
import com.hubspot.singularity.SingularityDeleteResult;
import com.hubspot.singularity.SingularityPendingDeploy;
import com.hubspot.singularity.SingularityTaskHistory;
import com.hubspot.singularity.SingularityTaskId;
import com.hubspot.singularity.config.SingularityConfiguration;
import com.hubspot.singularity.data.DeployManager;
import com.hubspot.singularity.data.TaskManager;

@Singleton
public class SingularityTaskHistoryPersister extends SingularityHistoryPersister<SingularityTaskId> {

  private static final Logger LOG = LoggerFactory.getLogger(SingularityTaskHistoryPersister.class);

  private final TaskManager taskManager;
  private final DeployManager deployManager;
  private final HistoryManager historyManager;

  @Inject
  public SingularityTaskHistoryPersister(SingularityConfiguration configuration, TaskManager taskManager, DeployManager deployManager, HistoryManager historyManager) {
    super(configuration);

    this.taskManager = taskManager;
    this.historyManager = historyManager;
    this.deployManager = deployManager;
  }

  @Override
  public void runActionOnPoll() {
    LOG.info("Checking inactive task ids for task history persistence");

    final long start = System.currentTimeMillis();

    final List<SingularityTaskId> allTaskIds = taskManager.getAllTaskIds();

    final Set<SingularityTaskId> activeTaskIds = Sets.newHashSet(taskManager.getActiveTaskIds());
    final Set<SingularityTaskId> lbCleaningTaskIds = Sets.newHashSet(taskManager.getLBCleanupTasks());
    final List<SingularityPendingDeploy> pendingDeploys = deployManager.getPendingDeploys();

    int numTotal = 0;
    int numTransferred = 0;

    for (SingularityTaskId taskId : allTaskIds) {
      if (activeTaskIds.contains(taskId) || lbCleaningTaskIds.contains(taskId) || isPartofPendingDeploy(pendingDeploys, taskId)) {
        continue;
      }

      final long age = start - taskId.getStartedAt();

      if (age < configuration.getTaskPersistAfterStartupBufferMillis()) {
        LOG.debug("Not persisting {}, it has started up too recently {} (buffer: {}) - this prevents race conditions with ZK tx", taskId, JavaUtils.durationFromMillis(age),
            JavaUtils.durationFromMillis(configuration.getTaskPersistAfterStartupBufferMillis()));
        continue;
      }

      if (moveToHistoryOrCheckForPurge(taskId)) {
        numTransferred++;
      }

      numTotal++;
    }

    LOG.info("Transferred {} out of {} inactive task ids (total {}) in {}", numTransferred, numTotal, allTaskIds.size(), JavaUtils.duration(start));
  }

  private boolean isPartofPendingDeploy(List<SingularityPendingDeploy> pendingDeploys, SingularityTaskId taskId) {
    for (SingularityPendingDeploy pendingDeploy : pendingDeploys) {
      if (pendingDeploy.getDeployMarker().getDeployId().equals(taskId.getDeployId()) && pendingDeploy.getDeployMarker().getRequestId().equals(taskId.getRequestId())) {
{'fraction_non_alphanumeric': 0.09304780576082039, 'fraction_numerical': 0.0012064545317448348, 'mean_word_length': 5.370797310278578, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7156313', 'n_tokens_mistral': 1929, 'n_tokens_neox': 1903, 'n_words': 367}