<commit_before>package eu.stratosphere.nephele.checkpointing;

import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import eu.stratosphere.nephele.execution.ExecutionObserver;
import eu.stratosphere.nephele.execution.ExecutionState;
import eu.stratosphere.nephele.executiongraph.ExecutionVertexID;
import eu.stratosphere.nephele.io.channels.ChannelID;
import eu.stratosphere.nephele.taskmanager.transferenvelope.CheckpointDeserializer;
import eu.stratosphere.nephele.taskmanager.transferenvelope.TransferEnvelope;
import eu.stratosphere.nephele.util.StringUtils;

final class ReplayThread extends Thread {

	private static final String REPLAY_SUFFIX = " (Replay)";

	private final ExecutionVertexID vertexID;

	private final ExecutionObserver executionObserver;

	private final boolean isCheckpointComplete;

	private final Map<ChannelID, ReplayOutputBroker> outputBrokerMap;

	/**
	 * The interval to sleep in case a communication channel is not yet entirely set up (in milliseconds).
	 */
	private static final int SLEEPINTERVAL = 100;

	private final AtomicBoolean restartRequested = new AtomicBoolean(false);

	ReplayThread(final ExecutionVertexID vertexID, final ExecutionObserver executionObserver, final String taskName,
			final boolean isCheckpointComplete, final Map<ChannelID, ReplayOutputBroker> outputBrokerMap) {
		super((taskName == null ? "Unkown" : taskName) + REPLAY_SUFFIX);

		this.vertexID = vertexID;
		this.executionObserver = executionObserver;
		this.isCheckpointComplete = isCheckpointComplete;
		this.outputBrokerMap = outputBrokerMap;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void run() {

		// Now the actual program starts to run
		changeExecutionState(ExecutionState.REPLAYING, null);

		// If the task has been canceled in the mean time, do not even start it
		if (this.executionObserver.isCanceled()) {
			changeExecutionState(ExecutionState.CANCELED, null);
			return;
		}

		try {

			replayCheckpoint();

			// Make sure, we enter the catch block when the task has been canceled
			if (this.executionObserver.isCanceled()) {
				throw new InterruptedException();
			}

		} catch (Exception e) {

			if (this.executionObserver.isCanceled()) {
				changeExecutionState(ExecutionState.CANCELED, null);
			} else {
				changeExecutionState(ExecutionState.FAILED, StringUtils.stringifyException(e));
			}

			return;
		}

		// Task finished running, but there may be some unconsumed data in the brokers
		changeExecutionState(ExecutionState.FINISHING, null);

		try {
			// Wait until all output broker have sent all of their data
			waitForAllOutputBrokerToFinish();
		} catch (Exception e) {

			if (this.executionObserver.isCanceled()) {
				changeExecutionState(ExecutionState.CANCELED, null);
			} else {
				changeExecutionState(ExecutionState.FAILED, StringUtils.stringifyException(e));
			}

			return;
		}

		// Finally, switch execution state to FINISHED and report to job manager
		changeExecutionState(ExecutionState.FINISHED, null);
	}

	private void waitForAllOutputBrokerToFinish() throws IOException, InterruptedException {

		while (true) {
			boolean finished = true;
			final Iterator<ReplayOutputBroker> it = this.outputBrokerMap.values().iterator();
			while (it.hasNext()) {

				if (it.next().hasFinished()) {
					finished = false;
				}
			}
<commit_msg>Fixed bug in ReplayThread close logic<commit_after>package eu.stratosphere.nephele.checkpointing;

import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import eu.stratosphere.nephele.execution.ExecutionObserver;
import eu.stratosphere.nephele.execution.ExecutionState;
import eu.stratosphere.nephele.executiongraph.ExecutionVertexID;
import eu.stratosphere.nephele.io.channels.ChannelID;
import eu.stratosphere.nephele.taskmanager.transferenvelope.CheckpointDeserializer;
import eu.stratosphere.nephele.taskmanager.transferenvelope.TransferEnvelope;
import eu.stratosphere.nephele.util.StringUtils;

final class ReplayThread extends Thread {

	private static final String REPLAY_SUFFIX = " (Replay)";

	private final ExecutionVertexID vertexID;

	private final ExecutionObserver executionObserver;

	private final boolean isCheckpointComplete;

	private final Map<ChannelID, ReplayOutputBroker> outputBrokerMap;

	/**
	 * The interval to sleep in case a communication channel is not yet entirely set up (in milliseconds).
	 */
	private static final int SLEEPINTERVAL = 100;

	private final AtomicBoolean restartRequested = new AtomicBoolean(false);

	ReplayThread(final ExecutionVertexID vertexID, final ExecutionObserver executionObserver, final String taskName,
			final boolean isCheckpointComplete, final Map<ChannelID, ReplayOutputBroker> outputBrokerMap) {
		super((taskName == null ? "Unkown" : taskName) + REPLAY_SUFFIX);

		this.vertexID = vertexID;
		this.executionObserver = executionObserver;
		this.isCheckpointComplete = isCheckpointComplete;
		this.outputBrokerMap = outputBrokerMap;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void run() {

		// Now the actual program starts to run
		changeExecutionState(ExecutionState.REPLAYING, null);

		// If the task has been canceled in the mean time, do not even start it
		if (this.executionObserver.isCanceled()) {
			changeExecutionState(ExecutionState.CANCELED, null);
			return;
		}

		try {

			replayCheckpoint();

			// Make sure, we enter the catch block when the task has been canceled
			if (this.executionObserver.isCanceled()) {
				throw new InterruptedException();
			}

		} catch (Exception e) {

			if (this.executionObserver.isCanceled()) {
				changeExecutionState(ExecutionState.CANCELED, null);
			} else {
				changeExecutionState(ExecutionState.FAILED, StringUtils.stringifyException(e));
			}

			return;
		}

		// Task finished running, but there may be some unconsumed data in the brokers
		changeExecutionState(ExecutionState.FINISHING, null);

		try {
			// Wait until all output broker have sent all of their data
			waitForAllOutputBrokerToFinish();
		} catch (Exception e) {

			if (this.executionObserver.isCanceled()) {
				changeExecutionState(ExecutionState.CANCELED, null);
			} else {
				changeExecutionState(ExecutionState.FAILED, StringUtils.stringifyException(e));
			}

			return;
		}

		// Finally, switch execution state to FINISHED and report to job manager
		changeExecutionState(ExecutionState.FINISHED, null);
	}

	private void waitForAllOutputBrokerToFinish() throws IOException, InterruptedException {

		while (true) {
			boolean finished = true;
			final Iterator<ReplayOutputBroker> it = this.outputBrokerMap.values().iterator();
			while (it.hasNext()) {

				if (!it.next().hasFinished()) {
					finished = false;
				}
			}

{'fraction_non_alphanumeric': 0.08820642266161828, 'fraction_numerical': 0.0008413967185527977, 'mean_word_length': 4.855500821018063, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '4560299', 'n_tokens_mistral': 2310, 'n_tokens_neox': 2029, 'n_words': 521}