<commit_before>/*
 * Copyright 2017 <NAME>
 *
 * 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 net.viktorc.pp4j.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import net.viktorc.pp4j.api.Command;
import net.viktorc.pp4j.api.DisruptedExecutionException;
import net.viktorc.pp4j.api.FailedCommandException;
import net.viktorc.pp4j.api.ProcessExecutorService;
import net.viktorc.pp4j.api.ProcessManagerFactory;
import net.viktorc.pp4j.api.Submission;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation of the {@link ProcessExecutorService} interface for maintaining and managing a pool of pre-started processes. The
 * processes are executed in instances of an own {@link net.viktorc.pp4j.api.ProcessExecutor} implementation. Each executor is assigned an
 * instance of an implementation of the {@link net.viktorc.pp4j.api.ProcessManager} interface using an implementation of the
 * {@link ProcessManagerFactory} interface. The pool accepts submissions in the form of {@link Submission} implementations which are
 * executed in any one of the available active process executors maintained by the pool. While executing a submission, the executor cannot
 * accept further submissions. The submissions are queued and executed as soon as there is an available executor. The size of the pool is
 * always kept between the minimum pool size and the maximum pool size (both inclusive). The reserve size specifies the minimum number of
 * processes that should always be available (there are no guarantees that there actually will be this many available executors at any
 * given time). This class uses <a href="https://www.slf4j.org/">SLF4J</a> for logging.
 *
 * @author <NAME>
 */
public class ProcessPoolExecutor implements ProcessExecutorService {

  private static final long DEFAULT_THREAD_KEEP_ALIVE_TIME = 60L * 1000L;
  private static final Logger LOGGER = LoggerFactory.getLogger(ProcessPoolExecutor.class);

  private final ProcessManagerFactory processManagerFactory;
  private final int minPoolSize;
  private final int maxPoolSize;
  private final int reserveSize;
  private final long threadKeepAliveTime;
  private final InternalProcessExecutorThreadPool processExecutorThreadPool;
  private final ExecutorService secondaryThreadPool;
  private final Queue<InternalProcessExecutor> processExecutors;
  private final BlockingDeque<InternalSubmission<?>> submissionQueue;
  private final CountDownLatch poolInitLatch;
  private final CountDownLatch poolTerminationLatch;
  private final Object mainLock;
  private final Object forceShutdownLock;

  private volatile int numOfSubmissions;
  private volatile boolean shutdown;

  /**
   * Constructs a pool of processes. The initial size of the pool is the minimum pool size or the reserve size depending on which one is
   * greater. This constructor blocks until the initial number of processes start up. The size of the pool is dynamically adjusted based on
   * the pool parameters and the rate of incoming submissions.
   *
   * @param processManagerFactory A {@link ProcessManagerFactory} instance that is used to build {@link net.viktorc.pp4j.api.ProcessManager}
   * instances that manage the processes' life cycle in the pool.
   * @param minPoolSize The minimum size of the process pool.
   * @param maxPoolSize The maximum size of the process pool.
   * @param reserveSize The number of available processes to keep in the pool.
   * @param threadKeepAliveTime The number of milliseconds of idleness after which threads are terminated if the sizes of the thread pools
   * exceed their core sizes.
   * @throws InterruptedException If the thread is interrupted while it is waiting for the core threads to start up.
   * @throws IllegalArgumentException If the manager factory is null, or the minimum pool size is less than 0, or the maximum pool size is
   * less than the minimum pool size or 1, or the reserve size is less than 0 or greater than the maximum pool size.
   */
  public ProcessPoolExecutor(ProcessManagerFactory processManagerFactory, int minPoolSize, int maxPoolSize, int reserveSize,
      long threadKeepAliveTime) throws InterruptedException {
    if (processManagerFactory == null) {
      throw new IllegalArgumentException("The process manager factory cannot be null");
    }
    if (minPoolSize < 0) {
      throw new IllegalArgumentException("The minimum pool size has to be greater than 0");
    }
    if (maxPoolSize < 1 || maxPoolSize < minPoolSize) {
      throw new IllegalArgumentException("The maximum pool size has to be at least 1 and at least as great as the minimum pool size");
    }
    if (reserveSize < 0 || reserveSize > maxPoolSize) {
      throw new IllegalArgumentException("The reserve has to be at least 0 and less than the maximum pool size");
    }
    if (threadKeepAliveTime <= 0) {
      throw new IllegalArgumentException("The thread keep-alive time must be greater than 0");
    }
    this.processManagerFactory = processManagerFactory;
    this.minPoolSize = minPoolSize;
    this.maxPoolSize = maxPoolSize;
    this.reserveSize = reserveSize;
    this.threadKeepAliveTime = threadKeepAliveTime;
    processExecutorThreadPool = new InternalProcessExecutorThreadPool();
    int actualMinSize = Math.max(minPoolSize, reserveSize);
    /* One normal process requires minimum 2 secondary threads (stdout listener, submission handler), 3 if
     * the stderr is not redirected to stdout (stderr listener), and 4 if keepAliveTime is positive (timer). */
    secondaryThreadPool = new ThreadPoolExecutor(2 * actualMinSize, Integer.MAX_VALUE, threadKeepAliveTime,
        TimeUnit.MILLISECONDS, new SynchronousQueue<>(), new CustomizedThreadFactory(this + "-secondaryThreadPool"));
    submissionQueue = new LinkedBlockingDeque<>();
    processExecutors = new LinkedBlockingQueue<>();
    poolInitLatch = new CountDownLatch(actualMinSize);
    poolTerminationLatch = new CountDownLatch(1);
    mainLock = new Object();
    forceShutdownLock = new Object();
    startPool(actualMinSize);
  }

  /**
   * Constructs a pool of processes. The initial size of the pool is the minimum pool size or the reserve size depending on which one is
   * greater. This constructor blocks until the initial number of processes start up. The size of the pool is dynamically adjusted based on
   * the pool parameters and the rate of incoming submissions.
   *
   * @param processManagerFactory A {@link ProcessManagerFactory} instance that is used to build {@link net.viktorc.pp4j.api.ProcessManager}
   * instances that manage the processes' life cycle in the pool.
   * @param minPoolSize The minimum size of the process pool.
   * @param maxPoolSize The maximum size of the process pool.
   * @param reserveSize The number of available processes to keep in the pool.
   * @throws InterruptedException If the thread is interrupted while it is waiting for the core threads to start up.
   * @throws IllegalArgumentException If the manager factory is null, or the minimum pool size is less than 0, or the maximum pool size is
   * less than the minimum pool size or 1, or the reserve size is less than 0 or greater than the maximum pool size.
   */
  public ProcessPoolExecutor(ProcessManagerFactory processManagerFactory, int minPoolSize, int maxPoolSize, int reserveSize)
      throws InterruptedException {
    this(processManagerFactory, minPoolSize, maxPoolSize, reserveSize, DEFAULT_THREAD_KEEP_ALIVE_TIME);
  }
<commit_msg>Ensure the internal thread pools are shut down if the constructor fails<commit_after>/*
 * Copyright 2017 <NAME>
 *
 * 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 net.viktorc.pp4j.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import net.viktorc.pp4j.api.Command;
import net.viktorc.pp4j.api.DisruptedExecutionException;
import net.viktorc.pp4j.api.FailedCommandException;
import net.viktorc.pp4j.api.ProcessExecutorService;
import net.viktorc.pp4j.api.ProcessManagerFactory;
import net.viktorc.pp4j.api.Submission;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation of the {@link ProcessExecutorService} interface for maintaining and managing a pool of pre-started processes. The
 * processes are executed in instances of an own {@link net.viktorc.pp4j.api.ProcessExecutor} implementation. Each executor is assigned an
 * instance of an implementation of the {@link net.viktorc.pp4j.api.ProcessManager} interface using an implementation of the
 * {@link ProcessManagerFactory} interface. The pool accepts submissions in the form of {@link Submission} implementations which are
 * executed in any one of the available active process executors maintained by the pool. While executing a submission, the executor cannot
 * accept further submissions. The submissions are queued and executed as soon as there is an available executor. The size of the pool is
 * always kept between the minimum pool size and the maximum pool size (both inclusive). The reserve size specifies the minimum number of
 * processes that should always be available (there are no guarantees that there actually will be this many available executors at any
 * given time). This class uses <a href="https://www.slf4j.org/">SLF4J</a> for logging.
 *
 * @author <NAME>
 */
public class ProcessPoolExecutor implements ProcessExecutorService {

  private static final long DEFAULT_THREAD_KEEP_ALIVE_TIME = 60L * 1000L;
  private static final Logger LOGGER = LoggerFactory.getLogger(ProcessPoolExecutor.class);

  private final ProcessManagerFactory processManagerFactory;
  private final int minPoolSize;
  private final int maxPoolSize;
  private final int reserveSize;
  private final long threadKeepAliveTime;
  private final InternalProcessExecutorThreadPool processExecutorThreadPool;
  private final ExecutorService secondaryThreadPool;
  private final Queue<InternalProcessExecutor> processExecutors;
  private final BlockingDeque<InternalSubmission<?>> submissionQueue;
  private final CountDownLatch poolInitLatch;
  private final CountDownLatch poolTerminationLatch;
  private final Object mainLock;
  private final Object forceShutdownLock;

  private volatile int numOfSubmissions;
  private volatile boolean shutdown;

  /**
   * Constructs a pool of processes. The initial size of the pool is the minimum pool size or the reserve size depending on which one is
   * greater. This constructor blocks until the initial number of processes start up. The size of the pool is dynamically adjusted based on
   * the pool parameters and the rate of incoming submissions.
   *
   * @param processManagerFactory A {@link ProcessManagerFactory} instance that is used to build {@link net.viktorc.pp4j.api.ProcessManager}
   * instances that manage the processes' life cycle in the pool.
   * @param minPoolSize The minimum size of the process pool.
   * @param maxPoolSize The maximum size of the process pool.
   * @param reserveSize The number of available processes to keep in the pool.
   * @param threadKeepAliveTime The number of milliseconds of idleness after which threads are terminated if the sizes of the thread pools
   * exceed their core sizes.
   * @throws InterruptedException If the thread is interrupted while it is waiting for the core threads to start up.
   * @throws IllegalArgumentException If the manager factory is null, or the minimum pool size is less than 0, or the maximum pool size is
   * less than the minimum pool size or 1, or the reserve size is less than 0 or greater than the maximum pool size.
   */
  public ProcessPoolExecutor(ProcessManagerFactory processManagerFactory, int minPoolSize, int maxPoolSize, int reserveSize,
      long threadKeepAliveTime) throws InterruptedException {
    if (processManagerFactory == null) {
      throw new IllegalArgumentException("The process manager factory cannot be null");
    }
    if (minPoolSize < 0) {
      throw new IllegalArgumentException("The minimum pool size has to be greater than 0");
    }
    if (maxPoolSize < 1 || maxPoolSize < minPoolSize) {
      throw new IllegalArgumentException("The maximum pool size has to be at least 1 and at least as great as the minimum pool size");
    }
    if (reserveSize < 0 || reserveSize > maxPoolSize) {
      throw new IllegalArgumentException("The reserve has to be at least 0 and less than the maximum pool size");
    }
    if (threadKeepAliveTime <= 0) {
      throw new IllegalArgumentException("The thread keep-alive time must be greater than 0");
    }
    this.processManagerFactory = processManagerFactory;
    this.minPoolSize = minPoolSize;
    this.maxPoolSize = maxPoolSize;
    this.reserveSize = reserveSize;
    this.threadKeepAliveTime = threadKeepAliveTime;
    processExecutorThreadPool = new InternalProcessExecutorThreadPool();
    int actualMinSize = Math.max(minPoolSize, reserveSize);
    /* One normal process requires minimum 2 secondary threads (stdout listener, submission handler), 3 if
     * the stderr is not redirected to stdout (stderr listener), and 4 if keepAliveTime is positive (timer). */
    secondaryThreadPool = new ThreadPoolExecutor(2 * actualMinSize, Integer.MAX_VALUE, threadKeepAliveTime,
        TimeUnit.MILLISECONDS, new SynchronousQueue<>(), new CustomizedThreadFactory(this + "-secondaryThreadPool"));
    submissionQueue = new LinkedBlockingDeque<>();
    processExecutors = new LinkedBlockingQueue<>();
    poolInitLatch = new CountDownLatch(actualMinSize);
    poolTerminationLatch = new CountDownLatch(1);
    mainLock = new Object();
    forceShutdownLock = new Object();
    try {
      startPool(actualMinSize);
    } catch (InterruptedException e) {
      processExecutorThreadPool.shutdownNow();
      secondaryThreadPool.shutdownNow();
      LOGGER.debug("Internal thread pools shut down due to constructor interruption");
      throw e;
    }
  }

  /**
   * Constructs a pool of processes. The initial size of the pool is the minimum pool size or the reserve size depending on which one is
   * greater. This constructor blocks until the initial number of processes start up. The size of the pool is dynamically adjusted based on
   * the pool parameters and the rate of incoming submissions.
   *
   * @param processManagerFactory A {@link ProcessManagerFactory} instance that is used to build {@link net.viktorc.pp4j.api.ProcessManager}
   * instances that manage the processes' life cycle in the pool.
   * @param minPoolSize The minimum size of the process pool.
   * @param maxPoolSize The maximum size of the process pool.
   * @param reserveSize The number of available processes to keep in the pool.
   * @throws InterruptedException If the thread is interrupted while it is waiting for the core threads to start up.
   * @throws IllegalArgumentException If the manager factory is null, or the minimum pool size is less than 0, or the maximum pool size is
   * less than the minimum pool size or 1, or the reserve size is less than 0 or greater than the maximum pool size.
   */
  public ProcessPoolExecutor(ProcessManagerFactory processManagerFactory, int minPoolSize, int maxPoolSize, int reserveSize)
      throws InterruptedException {
    this(processManagerFactory, minPoolSize, maxPoolSize, reserveSize, DEFAULT_THREAD_KEEP_ALIVE_TIME);
  }

{'fraction_non_alphanumeric': 0.061409904099601814, 'fraction_numerical': 0.005383882003252762, 'mean_word_length': 4.604022627278441, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 4, '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': '6887483', 'n_tokens_mistral': 4541, 'n_tokens_neox': 4434, 'n_words': 1924}