<commit_before>package cc.mallet.fst;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.logging.Logger;

import cc.mallet.types.InstanceList;
import cc.mallet.types.MatrixOps;

import cc.mallet.optimize.Optimizable;

import cc.mallet.util.MalletLogger;


/**
 * An adaptor for optimizables based on batch values/gradients.
 * <p>
 * Computes values, gradients for each batch in multiple threads and combines
 * them in the end.
 *
 * @author <NAME>
 * @see CRFOptimizableByBatchLabelLikelihood
 */
public class ThreadedOptimizable implements Optimizable.ByGradientValue {
	private static Logger logger = MalletLogger.getLogger(ThreadedOptimizable.class.getName());

	/** Data */
	protected InstanceList trainingSet;

	/** optimizable to be parallelized */
	protected Optimizable.ByCombiningBatchGradient optimizable;

	/** Value obtained from the optimizable for each batch */
	protected double[] batchCachedValue;
  /** Gradient obtained from the optimizable for each batch */
	protected List<double[]> batchCachedGradient;

	// determine when value/gradient become stale
	protected CacheStaleIndicator cacheIndicator;

	// tasks to be executed in individual threads, each task is instantiated only
	// once but executed in every iteration
	private transient Collection<Callable<Double>> valueTasks;
	private transient Collection<Callable<Boolean>> gradientTasks;

	// thread pool to compute value/gradient for one batch of data
	private transient ThreadPoolExecutor executor;

	// milliseconds
	public static final int SLEEP_TIME = 100;

	/**
	 * Initializes the optimizable and starts new threads.
	 *
	 * @param optimizable Optimizable to be parallelized
	 * @param numFactors Number of factors in model's parameters, used to
	 *        initialize the gradient
	 * @param cacheIndicator Determines when value/gradient become stale
	 */
	public ThreadedOptimizable(Optimizable.ByCombiningBatchGradient optimizable,
			InstanceList trainingSet, int numFactors,
			CacheStaleIndicator cacheIndicator) {
		// set up
		this.trainingSet = trainingSet;
		this.optimizable = optimizable;

		int numBatches = optimizable.getNumBatches();
		assert(numBatches > 0) : "Invalid number of batches: " + numBatches;
		batchCachedValue = new double[numBatches];
		batchCachedGradient = new ArrayList<double[]>(numBatches);
		for (int i = 0; i < numBatches; ++i) {
			batchCachedGradient.add(new double[numFactors]);
		}

		this.cacheIndicator = cacheIndicator;

		logger.info("Creating " + numBatches + " threads for updating gradient...");
		executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(numBatches);
		this.createTasks();
	}

	public Optimizable.ByCombiningBatchGradient getOptimizable() {
		return optimizable;
	}

	/**
	 * Shuts down the executor used to start and run threads to compute values
	 * and gradients.
	 * <p>
	 * *Note*: For a clean exit of all the threads, it is recommended to call
	 * this method after training finishes.
	 */
	public void shutdown() {
    assert(executor.shutdownNow().size() == 0) : "All tasks didn't finish";
	}

	public double getValue () {
		if (cacheIndicator.isValueStale()) {
			// compute values again
			try {
				// run all threads and wait for them to finish
				List<Future<Double>> results = executor.invokeAll(valueTasks);

				// compute final log probability
				int batch = 0;
				for (Future<Double> f : results) {
					try {
						batchCachedValue[batch++] = f.get();
					} catch (ExecutionException ee) {<commit_msg>Fixed hanging multi-threaded CRF code.<commit_after>package cc.mallet.fst;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

import cc.mallet.types.InstanceList;
import cc.mallet.types.MatrixOps;

import cc.mallet.optimize.Optimizable;

import cc.mallet.util.MalletLogger;


/**
 * An adaptor for optimizables based on batch values/gradients.
 * <p>
 * Computes values, gradients for each batch in multiple threads and combines
 * them in the end.
 *
 * @author <NAME>
 * @see CRFOptimizableByBatchLabelLikelihood
 */
public class ThreadedOptimizable implements Optimizable.ByGradientValue {
	private static Logger logger = MalletLogger.getLogger(ThreadedOptimizable.class.getName());

	/** Data */
	protected InstanceList trainingSet;

	/** optimizable to be parallelized */
	protected Optimizable.ByCombiningBatchGradient optimizable;

	/** Value obtained from the optimizable for each batch */
	protected double[] batchCachedValue;
  /** Gradient obtained from the optimizable for each batch */
	protected List<double[]> batchCachedGradient;

	// determine when value/gradient become stale
	protected CacheStaleIndicator cacheIndicator;

	// tasks to be executed in individual threads, each task is instantiated only
	// once but executed in every iteration
	private transient Collection<Callable<Double>> valueTasks;
	private transient Collection<Callable<Boolean>> gradientTasks;

	// thread pool to compute value/gradient for one batch of data
	private transient ThreadPoolExecutor executor;

	// milliseconds
	public static final int SLEEP_TIME = 100;

	/**
	 * Initializes the optimizable and starts new threads.
	 *
	 * @param optimizable Optimizable to be parallelized
	 * @param numFactors Number of factors in model's parameters, used to
	 *        initialize the gradient
	 * @param cacheIndicator Determines when value/gradient become stale
	 */
	public ThreadedOptimizable(Optimizable.ByCombiningBatchGradient optimizable,
			InstanceList trainingSet, int numFactors,
			CacheStaleIndicator cacheIndicator) {
		// set up
		this.trainingSet = trainingSet;
		this.optimizable = optimizable;

		int numBatches = optimizable.getNumBatches();
		assert(numBatches > 0) : "Invalid number of batches: " + numBatches;
		batchCachedValue = new double[numBatches];
		batchCachedGradient = new ArrayList<double[]>(numBatches);
		for (int i = 0; i < numBatches; ++i) {
			batchCachedGradient.add(new double[numFactors]);
		}

		this.cacheIndicator = cacheIndicator;

		logger.info("Creating " + numBatches + " threads for updating gradient...");
		executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(numBatches);
		this.createTasks();
	}

	public Optimizable.ByCombiningBatchGradient getOptimizable() {
		return optimizable;
	}

	/**
	 * Shuts down the executor used to start and run threads to compute values
	 * and gradients.
	 * <p>
	 * *Note*: For a clean exit of all the threads, it is recommended to call
	 * this method after training finishes.
	 */
	public void shutdown() {
		// fix submitted by <NAME> (<EMAIL>)
		executor.shutdown();
		try {
			executor.awaitTermination(30, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		assert(executor.shutdownNow().size() == 0) : "All tasks didn't finish";
	}

	public double getValue () {
		if (cacheIndicator.isValueStale()) {
			// compute values again
			try {
				// run all threads and wait for them to finish
				List<Future<Double>> results = executor.invokeAll(valueTasks);

				// compute final log probability
				int batch = 0;
				for (Future<Double> f : results) {
					try {
						batchCachedValue[batch++] = f.get();
					} catch (ExecutionException ee) {
{'fraction_non_alphanumeric': 0.09227579056505962, 'fraction_numerical': 0.002073613271124935, 'mean_word_length': 4.426863572433192, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, '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': '7475901', 'n_tokens_mistral': 2385, 'n_tokens_neox': 2140, 'n_words': 671}