<commit_before>package org.threadly.concurrent.wrapper.compatibility;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.threadly.concurrent.RunnableContainer;
import org.threadly.concurrent.SchedulerService;
import org.threadly.concurrent.future.FutureUtils;
import org.threadly.concurrent.future.ListenableFuture;
import org.threadly.concurrent.future.ListenableFutureTask;
import org.threadly.concurrent.future.ListenableScheduledFuture;
import org.threadly.util.ArgumentVerifier;
import org.threadly.util.Clock;
import org.threadly.util.ExceptionUtils;

/**
 * Generic implementation for the wrappers that implement {@link ScheduledExecutorService}.  This 
 * allows us to add new wrappers with the minimal amount of duplicated code.
 * 
 * @since 4.6.0 (since 2.0.0 at org.threadly.concurrent)
 */
abstract class AbstractExecutorServiceWrapper implements ScheduledExecutorService {
  protected final SchedulerService scheduler;
  
  /**
   * Constructs a new wrapper to adhere to the {@link ScheduledExecutorService} interface.
   * 
   * @param scheduler scheduler implementation to rely on
   */
  public AbstractExecutorServiceWrapper(SchedulerService scheduler) {
    ArgumentVerifier.assertNotNull(scheduler, "scheduler");
    
    this.scheduler = scheduler;
  }

  @Override
  public boolean isShutdown() {
    return scheduler.isShutdown();
  }

  @Override
  public <T> ListenableFuture<T> submit(Callable<T> task) {
    return scheduler.submit(task);
  }

  @Override
  public <T> ListenableFuture<T> submit(Runnable task, T result) {
    return scheduler.submit(task, result);
  }

  @Override
  public ListenableFuture<?> submit(Runnable task) {
    return scheduler.submit(task);
  }

  @Override
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
    return invokeAll(tasks, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
  }

  @Override
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                       long timeout, TimeUnit unit) throws InterruptedException {
    long timeoutInMs = unit.toMillis(timeout);
    long startTime = timeoutInMs < Long.MAX_VALUE ? Clock.accurateForwardProgressingMillis() : -1;
    List<Future<T>> resultList = new ArrayList<>(tasks.size());
    // execute all the tasks provided
    {
      Iterator<? extends Callable<T>> it = tasks.iterator();
      while (it.hasNext()) {
        Callable<T> c = it.next();
        if (c == null) {
          throw new NullPointerException();
        }
        
        ListenableFuture<T> lf = scheduler.submit(c);
        resultList.add(lf);
      }
    }
    // block till all tasks finish, or we reach our timeout
    if (timeoutInMs < Long.MAX_VALUE) {
      long remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime);
      try {
        FutureUtils.blockTillAllComplete(resultList, remainingTime);
      } catch (TimeoutException e) {
        FutureUtils.cancelIncompleteFutures(resultList, true);
      }
    } else {
      FutureUtils.blockTillAllComplete(resultList);
    }
    
    return resultList;
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException,
                                                                         ExecutionException {
    try {
      return invokeAny(tasks, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
      // basically impossible
      throw ExceptionUtils.makeRuntime(e);
    }
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks, 
                         long timeout, TimeUnit unit) throws InterruptedException,
                                                             ExecutionException, 
                                                             TimeoutException {
    if (tasks.size() < 1) {
      throw new IllegalArgumentException("Empty task list provided");
    }
    
    final long startTime = Clock.accurateForwardProgressingMillis();
    final long timeoutInMs = unit.toMillis(timeout);
    int failureCount = 0;
    // going to be optimistic and allocate the initialize size so that at most we have to do one expansion
    List<Future<T>> submittedFutures = new ArrayList<>((tasks.size() / 2) + 1);
    
    try {
      ExecutorCompletionService<T> ecs = new ExecutorCompletionService<>(this);
      ExecutionException lastEE = null;
      Iterator<? extends Callable<T>> it = tasks.iterator();
      // submit first one
      submittedFutures.add(ecs.submit(it.next()));

      long remainingTime;
      while (it.hasNext() && (remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime)) > 0) {
        Future<T> completedFuture = ecs.poll();
        if (completedFuture == null) {
          // submit another
          submittedFutures.add(ecs.submit(it.next()));
        } else {
          try {
            return completedFuture.get();
          } catch (ExecutionException e) {
            failureCount++;
            lastEE = e;
          }
        }
      }
      
      // we must compare against failure count otherwise we may throw a TimeoutException when all tasks have failed
      while ((remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime)) > 0 && 
             failureCount < submittedFutures.size()) {
        Future<T> completedFuture = ecs.poll(remainingTime, TimeUnit.MILLISECONDS);
        if (completedFuture == null) {
          throw new TimeoutException();<commit_msg>Fix for checkstyle warning<commit_after>package org.threadly.concurrent.wrapper.compatibility;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.threadly.concurrent.RunnableContainer;
import org.threadly.concurrent.SchedulerService;
import org.threadly.concurrent.future.FutureUtils;
import org.threadly.concurrent.future.ListenableFuture;
import org.threadly.concurrent.future.ListenableFutureTask;
import org.threadly.concurrent.future.ListenableScheduledFuture;
import org.threadly.util.ArgumentVerifier;
import org.threadly.util.Clock;
import org.threadly.util.ExceptionUtils;

/**
 * Generic implementation for the wrappers that implement {@link ScheduledExecutorService}.  This 
 * allows us to add new wrappers with the minimal amount of duplicated code.
 * 
 * @since 4.6.0 (since 2.0.0 at org.threadly.concurrent)
 */
abstract class AbstractExecutorServiceWrapper implements ScheduledExecutorService {
  protected final SchedulerService scheduler;
  
  /**
   * Constructs a new wrapper to adhere to the {@link ScheduledExecutorService} interface.
   * 
   * @param scheduler scheduler implementation to rely on
   */
  public AbstractExecutorServiceWrapper(SchedulerService scheduler) {
    ArgumentVerifier.assertNotNull(scheduler, "scheduler");
    
    this.scheduler = scheduler;
  }

  @Override
  public boolean isShutdown() {
    return scheduler.isShutdown();
  }

  @Override
  public <T> ListenableFuture<T> submit(Callable<T> task) {
    return scheduler.submit(task);
  }

  @Override
  public <T> ListenableFuture<T> submit(Runnable task, T result) {
    return scheduler.submit(task, result);
  }

  @Override
  public ListenableFuture<?> submit(Runnable task) {
    return scheduler.submit(task);
  }

  @Override
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
    return invokeAll(tasks, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
  }

  @Override
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                       long timeout, TimeUnit unit) throws InterruptedException {
    long timeoutInMs = unit.toMillis(timeout);
    long startTime = timeoutInMs < Long.MAX_VALUE ? Clock.accurateForwardProgressingMillis() : -1;
    List<Future<T>> resultList = new ArrayList<>(tasks.size());
    // execute all the tasks provided
    {
      Iterator<? extends Callable<T>> it = tasks.iterator();
      while (it.hasNext()) {
        Callable<T> c = it.next();
        if (c == null) {
          throw new NullPointerException();
        }
        
        ListenableFuture<T> lf = scheduler.submit(c);
        resultList.add(lf);
      }
    }
    // block till all tasks finish, or we reach our timeout
    if (timeoutInMs < Long.MAX_VALUE) {
      long remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime);
      try {
        FutureUtils.blockTillAllComplete(resultList, remainingTime);
      } catch (TimeoutException e) {
        FutureUtils.cancelIncompleteFutures(resultList, true);
      }
    } else {
      FutureUtils.blockTillAllComplete(resultList);
    }
    
    return resultList;
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException,
                                                                         ExecutionException {
    try {
      return invokeAny(tasks, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
      // basically impossible
      throw ExceptionUtils.makeRuntime(e);
    }
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks, 
                         long timeout, TimeUnit unit) throws InterruptedException,
                                                             ExecutionException, 
                                                             TimeoutException {
    if (tasks.size() < 1) {
      throw new IllegalArgumentException("Empty task list provided");
    }
    
    final long startTime = Clock.accurateForwardProgressingMillis();
    final long timeoutInMs = unit.toMillis(timeout);
    int failureCount = 0;
    // going to be optimistic and allocate the initialize size so that at most we have to do one expansion
    List<Future<T>> submittedFutures = new ArrayList<>((tasks.size() / 2) + 1);
    
    try {
      ExecutorCompletionService<T> ecs = new ExecutorCompletionService<>(this);
      ExecutionException lastEE = null;
      Iterator<? extends Callable<T>> it = tasks.iterator();
      // submit first one
      submittedFutures.add(ecs.submit(it.next()));

      long remainingTime;
      while (it.hasNext() && 
             (remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime)) > 0) {
        Future<T> completedFuture = ecs.poll();
        if (completedFuture == null) {
          // submit another
          submittedFutures.add(ecs.submit(it.next()));
        } else {
          try {
            return completedFuture.get();
          } catch (ExecutionException e) {
            failureCount++;
            lastEE = e;
          }
        }
      }
      
      // we must compare against failure count otherwise we may throw a TimeoutException when all tasks have failed
      while ((remainingTime = timeoutInMs - (Clock.lastKnownForwardProgressingMillis() - startTime)) > 0 && 
             failureCount < submittedFutures.size()) {
        Future<T> completedFuture = ecs.poll(remainingTime, TimeUnit.MILLISECONDS);
        if (completedFuture == null) {
          throw new TimeoutException();
{'fraction_non_alphanumeric': 0.09643605870020965, 'fraction_numerical': 0.002180293501048218, 'mean_word_length': 2.930784442979565, 'pattern_counts': {'":': 0, '<': 87, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4503882', 'n_tokens_mistral': 3228, 'n_tokens_neox': 3164, 'n_words': 796}