<commit_before>package nl.vu.cs.ajira.datalayer.chainsplits;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import nl.vu.cs.ajira.Context;
import nl.vu.cs.ajira.actions.ActionContext;
import nl.vu.cs.ajira.actions.ActionOutput;
import nl.vu.cs.ajira.actions.ActionSequence;
import nl.vu.cs.ajira.chains.ChainLocation;
import nl.vu.cs.ajira.data.types.SimpleData;
import nl.vu.cs.ajira.data.types.TInt;
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ChainSplitLayer extends InputLayer {

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

	public static class SplitIterator extends TupleIterator implements
			ActionOutput {

		private final int id;
		private boolean isOpen = true;
		private Tuple tuple = null;

		/**
		 * Custom constructor.
		 * 
		 * @param id
		 *            The id of the SplitIterator.
		 */
		SplitIterator(int id) {
			this.id = id;
		}

		/**
		 * 
		 * @return The id of the SplitIterator
		 */
		public int getId() {
			return id;
		}

		@Override
		public synchronized boolean next() {
			while (isOpen && tuple == null) {
				try {
					wait();
				} catch (InterruptedException e) {
					// ignore
				}
			}
			return isOpen;
		}

		@Override
		public synchronized void getTuple(Tuple tuple) {
			this.tuple.copyTo(tuple);
			this.tuple = null;
			notify();
		}

		@Override
		public boolean isReady() {
			return true;
		}

		@Override
		public synchronized void output(Tuple tuple) {
			while (this.tuple != null) {
				try {
					wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			this.tuple = tuple;
			notify();
		}

		public synchronized void close() {
			isOpen = false;
			this.notify();
		}

		@Override
		public void branch(ActionSequence actions) throws Exception {
			throw new Exception("Not allowed");
		}

		@Override
		public ActionOutput split(int reconnectAt, ActionSequence actions)
				throws Exception {
			throw new Exception("Not allowed");
		}

		// FIXME: Very very inefficient. Need to fix it!
		@Override
		public synchronized void output(SimpleData... data) {
			while (this.tuple != null) {
				try {
					wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			this.tuple = TupleFactory.newTuple(data);
			notify();
		}
	}

	private static final ChainSplitLayer instance = new ChainSplitLayer();
	public Map<Integer, SplitIterator> existingSplits = new ConcurrentHashMap<Integer, SplitIterator>();
	private int counter = 0;

	private ChainSplitLayer() {
	}

	public static ChainSplitLayer getInstance() {
		return instance;
	}
<commit_msg>previous fix was wrong. Original code re-instated.<commit_after>package nl.vu.cs.ajira.datalayer.chainsplits;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import nl.vu.cs.ajira.Context;
import nl.vu.cs.ajira.actions.ActionContext;
import nl.vu.cs.ajira.actions.ActionOutput;
import nl.vu.cs.ajira.actions.ActionSequence;
import nl.vu.cs.ajira.chains.ChainLocation;
import nl.vu.cs.ajira.data.types.SimpleData;
import nl.vu.cs.ajira.data.types.TInt;
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ChainSplitLayer extends InputLayer {

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

	public static class SplitIterator extends TupleIterator implements
			ActionOutput {

		private final int id;
		private boolean isOpen = true;
		private Tuple tuple = null;

		/**
		 * Custom constructor.
		 * 
		 * @param id
		 *            The id of the SplitIterator.
		 */
		SplitIterator(int id) {
			this.id = id;
		}

		/**
		 * 
		 * @return The id of the SplitIterator
		 */
		public int getId() {
			return id;
		}

		@Override
		public synchronized boolean next() {
			while (isOpen && tuple == null) {
				try {
					wait();
				} catch (InterruptedException e) {
					// ignore
				}
			}
			return isOpen;
		}

		@Override
		public synchronized void getTuple(Tuple tuple) {
			this.tuple.copyTo(tuple);
			this.tuple = null;
			notify();
		}

		@Override
		public boolean isReady() {
			return true;
		}

		@Override
		public synchronized void output(Tuple tuple) {
			this.tuple = tuple;
			notify();
			try {
				wait();
			} catch (InterruptedException e) {
				// ignore
			}
		}

		public synchronized void close() {
			isOpen = false;
			this.notify();
		}

		@Override
		public void branch(ActionSequence actions) throws Exception {
			throw new Exception("Not allowed");
		}

		@Override
		public ActionOutput split(int reconnectAt, ActionSequence actions)
				throws Exception {
			throw new Exception("Not allowed");
		}

		// FIXME: Very very inefficient. Need to fix it!
		@Override
		public synchronized void output(SimpleData... data) {
			this.tuple = TupleFactory.newTuple(data);
			notify();
			try {
				wait();
			} catch (InterruptedException e) {
				// ignore
			}
		}
	}

	private static final ChainSplitLayer instance = new ChainSplitLayer();
	public Map<Integer, SplitIterator> existingSplits = new ConcurrentHashMap<Integer, SplitIterator>();
	private int counter = 0;

	private ChainSplitLayer() {
	}

	public static ChainSplitLayer getInstance() {
		return instance;
	}

{'fraction_non_alphanumeric': 0.1091455864142933, 'fraction_numerical': 0.0010613833362816203, 'mean_word_length': 3.251127819548872, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '14116', 'n_tokens_mistral': 2103, 'n_tokens_neox': 1776, 'n_words': 448}