<commit_before>package nl.vu.cs.ajira.chains;

import java.util.List;
import java.util.Map;

import nl.vu.cs.ajira.Context;
import nl.vu.cs.ajira.actions.ActionFactory;
import nl.vu.cs.ajira.actions.support.Query;
import nl.vu.cs.ajira.buckets.Bucket;
import nl.vu.cs.ajira.buckets.BucketIterator;
import nl.vu.cs.ajira.data.types.Tuple;
import nl.vu.cs.ajira.data.types.TupleFactory;
import nl.vu.cs.ajira.datalayer.InputLayer;
import nl.vu.cs.ajira.datalayer.TupleIterator;
import nl.vu.cs.ajira.mgmt.StatisticsCollector;
import nl.vu.cs.ajira.net.NetworkLayer;
import nl.vu.cs.ajira.storage.containers.WritableContainer;
import nl.vu.cs.ajira.utils.Consts;

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

public class ChainHandler implements Runnable {

	static final Logger log = LoggerFactory.getLogger(ChainHandler.class);

	public static final int STATUS_INACTIVE = 0;
	public static final int STATUS_ACTIVE = 1;
	public static final int STATUS_WAIT = 2;
	public static final int STATUS_FINISHED = 3;

	private Context context = null;
	private NetworkLayer net = null;
	private WritableContainer<Chain> chainsToProcess = null;
	private ActionFactory ap = null;
	private StatisticsCollector stats = null;
	private Chain currentChain = new Chain();

	private boolean submissionFailed;
	private int status = STATUS_INACTIVE;
	public boolean singleChain = false;
	private final Query query = new Query();
	private final Tuple tuple = TupleFactory.newTuple();
	private ChainExecutor actions;

	ChainHandler(Context context) {
		this.context = context;
		this.net = context.getNetworkLayer();
		this.chainsToProcess = context.getChainHandlerManager()
				.getChainsToProcess();
		this.stats = context.getStatisticsCollector();
		this.ap = context.getActionsProvider();
		actions = new ChainExecutor(this, context);
	}

	ChainHandler(Context context, Chain chain) {
		this(context);
		singleChain = true;
		chain.copyTo(this.currentChain);
	}

	private void processChain() {

		status = STATUS_ACTIVE;
		// Init the environment
		actions.init(currentChain);

		try {
			currentChain.getActions(actions, ap);
		} catch (Throwable e) {
			// Broadcast all the nodes that a chain part of a job has
			// failed.
			log.error("getActions() on chain failed, cancelling the job ...", e);
			try {
				net.signalChainFailed(currentChain, e);
			} catch (Throwable e1) {
				log.error("Failed in managing to cancel the job", e);
			}
			status = STATUS_INACTIVE;
			return;
		}

		if (actions.getNActions() > 0) {

			// Read the input tuple from the knowledge base
			currentChain.getQuery(query);
			int il = currentChain.getInputLayer();
			InputLayer input = context.getInputLayer(il);
			TupleIterator itr = input.getIterator(query.getTuple(), actions);
			if (!itr.isReady()) {
				context.getChainNotifier().addWaiter(itr, currentChain);
				currentChain = new Chain();
				status = STATUS_INACTIVE;
				return;
			}

			if (il == Consts.BUCKET_INPUT_LAYER_ID) {
				// Check whether there are some counters that should be added to
				// the ChainExecutor
				BucketIterator bi = (BucketIterator) itr;
				Bucket b = bi.getBucket();
				Map<Long, List<Integer>> counters = b
						.getAdditionalChildrenCounts();
				if (counters != null && counters.size() > 0) {
					// Add them to the ChainExecutor
					actions.addAndUpdateCounters(counters);
				}
			}

			if (log.isDebugEnabled()) {
				log.debug("Starting chain " + currentChain.getChainId());
			}

			submissionFailed = false;

			/***** START CHAIN *****/
			long timeCycle = System.currentTimeMillis();
			actions.setInputIterator(itr);
			try {
				actions.startProcess();

				// Process the data on the chain
				boolean eof;
				do {
					eof = !itr.nextTuple();
					if (!eof) {
						itr.getTuple(tuple);
						actions.output(tuple);
					} else { // EOF Case
						actions.stopProcess();
					}
				} while (!eof && !getSubmissionFailed());

				if (log.isDebugEnabled()) {
					timeCycle = System.currentTimeMillis() - timeCycle;
					log.debug("Chain " + currentChain.getChainId()
							+ "runtime cycle: " + timeCycle);
				}

				stats.addCounter(currentChain.getSubmissionNode(),
						currentChain.getSubmissionId(), "Chains Processed", 1);

			} catch (Throwable e) {
				// Broadcast all the nodes that a chain part of a job has
				// failed.
				log.error("Chain failed. Cancelling the job ...", e);
				try {
					net.signalChainFailed(currentChain, e);
				} catch (Exception e1) {
					log.error("Failed in managing to cancel the job.", e);
				}
			} finally {
				input.releaseIterator(itr, actions);
				status = STATUS_INACTIVE;
			}
		}
	}

	private boolean getSubmissionFailed() {
		return submissionFailed;
	}

	public void submissionFailed(int submissionId) {
		if (currentChain != null
				&& currentChain.getSubmissionId() == submissionId) {
			submissionFailed = true;
		}
	}

	@Override
	public void run() {

		if (singleChain) {
			if (log.isDebugEnabled()) {
				log.debug("Single chain handler");
			}
			processChain();
			if (log.isDebugEnabled()) {
				log.debug("Single chain handler done");
			}
			status = STATUS_FINISHED;
			return;
		}

		while (true) {
			// Get a new chain to process
			chainsToProcess.remove(currentChain);
			processChain();
		}
	}

	void setStatus(int status) {
		this.status = status;
	}

	public int getStatus() {
		return status;
	}
}<commit_msg>Improvement in error handling<commit_after>package nl.vu.cs.ajira.chains;

import java.util.List;
import java.util.Map;

import nl.vu.cs.ajira.Context;
import nl.vu.cs.ajira.actions.ActionFactory;
import nl.vu.cs.ajira.actions.support.Query;
import nl.vu.cs.ajira.buckets.Bucket;
import nl.vu.cs.ajira.buckets.BucketIterator;
import nl.vu.cs.ajira.data.types.Tuple;
import nl.vu.cs.ajira.data.types.TupleFactory;
import nl.vu.cs.ajira.datalayer.InputLayer;
import nl.vu.cs.ajira.datalayer.TupleIterator;
import nl.vu.cs.ajira.mgmt.StatisticsCollector;
import nl.vu.cs.ajira.net.NetworkLayer;
import nl.vu.cs.ajira.storage.containers.WritableContainer;
import nl.vu.cs.ajira.utils.Consts;

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

public class ChainHandler implements Runnable {

	static final Logger log = LoggerFactory.getLogger(ChainHandler.class);

	public static final int STATUS_INACTIVE = 0;
	public static final int STATUS_ACTIVE = 1;
	public static final int STATUS_WAIT = 2;
	public static final int STATUS_FINISHED = 3;

	private Context context = null;
	private NetworkLayer net = null;
	private WritableContainer<Chain> chainsToProcess = null;
	private ActionFactory ap = null;
	private StatisticsCollector stats = null;
	private Chain currentChain = new Chain();

	private boolean submissionFailed;
	private int status = STATUS_INACTIVE;
	public boolean singleChain = false;
	private final Query query = new Query();
	private final Tuple tuple = TupleFactory.newTuple();
	private ChainExecutor actions;

	ChainHandler(Context context) {
		this.context = context;
		this.net = context.getNetworkLayer();
		this.chainsToProcess = context.getChainHandlerManager()
				.getChainsToProcess();
		this.stats = context.getStatisticsCollector();
		this.ap = context.getActionsProvider();
		actions = new ChainExecutor(this, context);
	}

	ChainHandler(Context context, Chain chain) {
		this(context);
		singleChain = true;
		chain.copyTo(this.currentChain);
	}

	private void processChain() throws Exception {

		// Init the environment
		actions.init(currentChain);

		currentChain.getActions(actions, ap);

		if (actions.getNActions() > 0) {

			// Read the input tuple from the knowledge base
			currentChain.getQuery(query);
			int il = currentChain.getInputLayer();
			InputLayer input = context.getInputLayer(il);
			TupleIterator itr = input.getIterator(query.getTuple(), actions);
			if (!itr.isReady()) {
				context.getChainNotifier().addWaiter(itr, currentChain);
				currentChain = new Chain();
				return;
			}

			if (il == Consts.BUCKET_INPUT_LAYER_ID) {
				// Check whether there are some counters that should be added to
				// the ChainExecutor
				BucketIterator bi = (BucketIterator) itr;
				Bucket b = bi.getBucket();
				Map<Long, List<Integer>> counters = b
						.getAdditionalChildrenCounts();
				if (counters != null && counters.size() > 0) {
					// Add them to the ChainExecutor
					actions.addAndUpdateCounters(counters);
				}
			}

			if (log.isDebugEnabled()) {
				log.debug("Starting chain " + currentChain.getChainId());
			}

			submissionFailed = false;

			/***** START CHAIN *****/
			long timeCycle = System.currentTimeMillis();
			actions.setInputIterator(itr);
			try {
				actions.startProcess();

				// Process the data on the chain
				boolean eof;
				do {
					eof = !itr.nextTuple();
					if (!eof) {
						itr.getTuple(tuple);
						actions.output(tuple);
					} else { // EOF Case
						actions.stopProcess();
					}
				} while (!eof && !getSubmissionFailed());

				if (log.isDebugEnabled()) {
					timeCycle = System.currentTimeMillis() - timeCycle;
					log.debug("Chain " + currentChain.getChainId()
							+ "runtime cycle: " + timeCycle);
				}

				stats.addCounter(currentChain.getSubmissionNode(),
						currentChain.getSubmissionId(), "Chains Processed", 1);

			} finally {
				input.releaseIterator(itr, actions);
			}
		}
	}

	private boolean getSubmissionFailed() {
		return submissionFailed;
	}

	public void submissionFailed(int submissionId) {
		if (currentChain != null
				&& currentChain.getSubmissionId() == submissionId) {
			submissionFailed = true;
		}
	}

	@Override
	public void run() {

		if (singleChain) {
			if (log.isDebugEnabled()) {
				log.debug("Single chain handler");
			}
			status = STATUS_ACTIVE;
			try {
				processChain();
			} catch(Throwable e) {
				// Broadcast all the nodes that a chain part of a job has
				// failed.
				log.error("chain failed, cancelling the job ...", e);
				try {
					net.signalChainFailed(currentChain, e);
				} catch (Throwable e1) {
					log.error("Failed in managing to cancel the job", e);
				}
			}
			if (log.isDebugEnabled()) {
				log.debug("Single chain handler done");
			}
			status = STATUS_FINISHED;
			return;
		}

		while (true) {
			// Get a new chain to process
			chainsToProcess.remove(currentChain);
			try {
				status = STATUS_ACTIVE;
				processChain();
			} catch (Throwable e) {
				// Broadcast all the nodes that a chain part of a job has
				// failed.
				log.error("Chain failed. Cancelling the job ...", e);
				try {
					net.signalChainFailed(currentChain, e);
				} catch (Exception e1) {
					log.error("Failed in managing to cancel the job.", e);
				}
			} finally {
				status = STATUS_INACTIVE;
			}
		}
	}

	void setStatus(int status) {
		this.status = status;
	}

	public int getStatus() {
		return status;
	}
}
{'fraction_non_alphanumeric': 0.11772878844016302, 'fraction_numerical': 0.002037791774731382, 'mean_word_length': 3.4968763015410245, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '6530273', 'n_tokens_mistral': 4046, 'n_tokens_neox': 3408, 'n_words': 839}