<commit_before>package arch.actions.files;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;

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

import arch.ActionContext;
import arch.actions.Action;
import arch.chains.Chain;
import arch.data.types.TInt;
import arch.data.types.TString;
import arch.data.types.Tuple;
import arch.datalayer.files.FileCollection;
import arch.datalayer.files.FilesLayer;
import arch.storage.container.WritableContainer;
import arch.utils.Consts;

public class FileSplitter extends Action {

	public static final String MINIMUM_SPLIT_SIZE = "splitinput.minimumsize";
	public static final int MINIMUM_FILE_SPLIT = 4 * 1024 * 1024; // 1 MB

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

	private final Tuple tuple = new Tuple();
	private int minimumFileSplitSize;
	private FileCollection currentFileSplit;
	private int splitId;

	@Override
	public void readFrom(DataInput input) throws IOException {
	}

	@Override
	public void writeTo(DataOutput output) throws IOException {
	}

	@Override
	public int bytesToStore() {
		return 0;
	}

	private Chain processSplit(ActionContext context, Chain chain,
			WritableContainer<Chain> chainsToProcess) throws Exception {
		String key = "split-" + splitId++;
		context.putObjectInCache(key, currentFileSplit);

		Chain newChain = new Chain();
		chain.createBranch(context, newChain);
		tuple.set(new TInt(FilesLayer.OP_READ), new TString(key), new TInt(
				context.getNetworkLayer().getMyPartition()));
		newChain.replaceInputTuple(tuple);
		newChain.setInputLayerId(Consts.DEFAULT_INPUT_LAYER_ID);

		chainsToProcess.add(newChain);
		currentFileSplit = new FileCollection();

		return newChain;
	}

	@Override
	public void startProcess(ActionContext context, Chain chain)
			throws Exception {
		minimumFileSplitSize = context.getConfiguration().getInt(
				MINIMUM_SPLIT_SIZE, MINIMUM_FILE_SPLIT);
		currentFileSplit = new FileCollection();
		splitId = 0;
	}

	@Override
	public void process(Tuple inputTuple, Chain remainingChain,
			WritableContainer<Chain> chainsToResolve,
			WritableContainer<Chain> chainsToProcess,
			WritableContainer<Tuple> output, ActionContext context)
			throws Exception {

		// In input I receive a list of files
		TString path = new TString();
		inputTuple.get(path, 0);
		File file = new File(path.getValue());

		long sizeFile = file.length();
		if (currentFileSplit.getSize() + sizeFile >= minimumFileSplitSize) {
			processSplit(context, remainingChain, chainsToProcess);
		} else {
			currentFileSplit.addFile(file);
		}
	}

	@Override
	public void stopProcess(ActionContext context, Chain chain,
			WritableContainer<Tuple> output,
			WritableContainer<Chain> newChains,
			WritableContainer<Chain> chainsToSend) throws Exception {
		if (currentFileSplit.getSize() > 0) {
			processSplit(context, chain, newChains);
		}
	}
}
<commit_msg>bugfix<commit_after>package arch.actions.files;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;

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

import arch.ActionContext;
import arch.actions.Action;
import arch.chains.Chain;
import arch.data.types.TInt;
import arch.data.types.TString;
import arch.data.types.Tuple;
import arch.datalayer.files.FileCollection;
import arch.datalayer.files.FilesLayer;
import arch.storage.container.WritableContainer;
import arch.utils.Consts;

public class FileSplitter extends Action {

	public static final String MINIMUM_SPLIT_SIZE = "splitinput.minimumsize";
	public static final int MINIMUM_FILE_SPLIT = 4 * 1024 * 1024; // 1 MB

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

	private final Tuple tuple = new Tuple();
	private int minimumFileSplitSize;
	private FileCollection currentFileSplit;
	private int splitId;

	@Override
	public void readFrom(DataInput input) throws IOException {
	}

	@Override
	public void writeTo(DataOutput output) throws IOException {
	}

	@Override
	public int bytesToStore() {
		return 0;
	}

	private Chain processSplit(ActionContext context, Chain chain,
			WritableContainer<Chain> chainsToProcess) throws Exception {
		String key = "split-" + splitId++;
		context.putObjectInCache(key, currentFileSplit);

		Chain newChain = new Chain();
		chain.createBranch(context, newChain);
		tuple.set(new TInt(FilesLayer.OP_READ), new TString(key), new TInt(
				context.getNetworkLayer().getMyPartition()));
		newChain.replaceInputTuple(tuple);
		newChain.setInputLayerId(Consts.DEFAULT_INPUT_LAYER_ID);

		chainsToProcess.add(newChain);
		currentFileSplit = new FileCollection();

		return newChain;
	}

	@Override
	public void startProcess(ActionContext context, Chain chain)
			throws Exception {
		minimumFileSplitSize = context.getConfiguration().getInt(
				MINIMUM_SPLIT_SIZE, MINIMUM_FILE_SPLIT);
		currentFileSplit = new FileCollection();
		splitId = 0;
	}

	@Override
	public void process(Tuple inputTuple, Chain remainingChain,
			WritableContainer<Chain> chainsToResolve,
			WritableContainer<Chain> chainsToProcess,
			WritableContainer<Tuple> output, ActionContext context)
			throws Exception {

		// In input I receive a list of files
		TString path = new TString();
		inputTuple.get(path, 0);
		File file = new File(path.getValue());

		long sizeFile = file.length();
		if (currentFileSplit.getSize() + sizeFile >= minimumFileSplitSize) {
			processSplit(context, remainingChain, chainsToProcess);
		}
		currentFileSplit.addFile(file);
	}

	@Override
	public void stopProcess(ActionContext context, Chain chain,
			WritableContainer<Tuple> output,
			WritableContainer<Chain> newChains,
			WritableContainer<Chain> chainsToSend) throws Exception {
		if (currentFileSplit.getSize() > 0) {
			processSplit(context, chain, newChains);
		}
		context.incrCounter("# file splits", splitId);
	}
}

{'fraction_non_alphanumeric': 0.09494743981010512, 'fraction_numerical': 0.0054255679891488644, 'mean_word_length': 5.343010752688172, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, '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': '5272906', 'n_tokens_mistral': 1939, 'n_tokens_neox': 1768, 'n_words': 439}