<commit_before>package edu.mit.streamjit.impl.common;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import edu.mit.streamjit.api.CompiledStream;
import edu.mit.streamjit.api.OneToOneElement;
import edu.mit.streamjit.api.StreamCompiler;
import edu.mit.streamjit.api.Worker;
import edu.mit.streamjit.impl.blob.Blob;
import edu.mit.streamjit.impl.blob.Blob.Token;
import edu.mit.streamjit.impl.blob.BlobFactory;
import edu.mit.streamjit.impl.blob.Buffer;
import edu.mit.streamjit.impl.blob.Buffers;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * A StreamCompiler that uses a BlobFactory to make a Blob for the entire graph.
 *
 * TODO: there should be a way to specify a custom configuration, but the blob's
 * default config depends on the workers (e.g., the compiler has parameters for
 * each worker to decide if that worker should be fused).  We'll need some
 * notion of a configuration override, or possibly split
 * Blob.getDefaultConfiguration() into worker-specific and worker-independent
 * parts.
 *
 * TODO: should this be in the blob package?
 * @author <NAME> <<EMAIL>>
 * @since 7/27/2013
 */
public class BlobHostStreamCompiler implements StreamCompiler {
	private final BlobFactory blobFactory;
	private int maxNumCores;
	public BlobHostStreamCompiler(BlobFactory blobFactory, int maxNumCores) {
		this.blobFactory = blobFactory;
		this.maxNumCores = maxNumCores;
	}

	/**
	 * Set the maximum number of cores that will be used to compile.
	 *
	 * Subclasses exposing other options should override this method to return
	 * their type.  (Java doesn't have self types, and CRTP doesn't seem right
	 * here.)
	 * @param maxNumCores the maximum number of cores to use
	 * @return this, for chaining
	 */
	public BlobHostStreamCompiler maxNumCores(int maxNumCores) {
		this.maxNumCores = maxNumCores;
		return this;
	}

	@Override
	public <I, O> CompiledStream<I, O> compile(OneToOneElement<I, O> stream) {
		ConnectWorkersVisitor cwv = new ConnectWorkersVisitor();
		stream.visit(cwv);
		ImmutableSet<Worker<?, ?>> workers = Workers.getAllWorkersInGraph(cwv.getSource());
		Blob blob = blobFactory.makeBlob(workers, getConfiguration(workers), maxNumCores);

		Token inputToken = Iterables.getOnlyElement(blob.getInputs());
		Token outputToken = Iterables.getOnlyElement(blob.getOutputs());
		Buffer inputBuffer = Buffers.blockingQueueBuffer(new ArrayBlockingQueue<>(blob.getMinimumBufferCapacity(inputToken)), false, false);
		Buffer outputBuffer = Buffers.blockingQueueBuffer(new ArrayBlockingQueue<>(blob.getMinimumBufferCapacity(outputToken)), false, false);
		ImmutableMap<Token, Buffer> bufferMap = ImmutableMap.<Token, Buffer>builder()
				.put(inputToken, inputBuffer)
				.put(outputToken, outputBuffer)
				.build();
		blob.installBuffers(bufferMap);

		ImmutableList.Builder<PollingCoreThread> threads = ImmutableList.builder();
		for (int i = 0; i < blob.getCoreCount(); ++i) {
			PollingCoreThread thread = new PollingCoreThread(blob.getCoreCode(i), blob.toString()+"-"+i);
			threads.add(thread);
			thread.start();
		}

		return new BlobHostCompiledStream<>(blob, inputBuffer, outputBuffer, threads.build());
	}

	/**
	 * Get a configuration for the given worker.  This implementation returns
	 * the default configuration from the blob factory, but subclasses can
	 * override this to implement compiler-specific options.
	 * @param workers the set of workers in the blob
	 * @return a configuration
	 */
	protected Configuration getConfiguration(Set<Worker<?, ?>> workers) {
		return blobFactory.getDefaultConfiguration(workers);
	}

	private static final class BlobHostCompiledStream<I, O> implements CompiledStream<I, O> {
		private final Blob blob;
		private final Buffer inputBuffer, outputBuffer;
		private final ImmutableList<PollingCoreThread> threads;
		private BlobHostCompiledStream(Blob blob, Buffer inputBuffer, Buffer outputBuffer, ImmutableList<PollingCoreThread> threads) {
			this.blob = blob;
			this.inputBuffer = inputBuffer;
			this.outputBuffer = outputBuffer;
			this.threads = threads;
		}

		@Override
		public boolean offer(I input) {
			return inputBuffer.write(input);
		}

		@Override
		@SuppressWarnings("unchecked")
		public O poll() {
			return (O)outputBuffer.read();
		}<commit_msg>BlobHostStreamCompiler: actually, let's do that with a hook<commit_after>package edu.mit.streamjit.impl.common;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import edu.mit.streamjit.api.CompiledStream;
import edu.mit.streamjit.api.OneToOneElement;
import edu.mit.streamjit.api.StreamCompiler;
import edu.mit.streamjit.api.Worker;
import edu.mit.streamjit.impl.blob.Blob;
import edu.mit.streamjit.impl.blob.Blob.Token;
import edu.mit.streamjit.impl.blob.BlobFactory;
import edu.mit.streamjit.impl.blob.Buffer;
import edu.mit.streamjit.impl.blob.Buffers;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * A StreamCompiler that uses a BlobFactory to make a Blob for the entire graph.
 *
 * TODO: there should be a way to specify a custom configuration, but the blob's
 * default config depends on the workers (e.g., the compiler has parameters for
 * each worker to decide if that worker should be fused).  We'll need some
 * notion of a configuration override, or possibly split
 * Blob.getDefaultConfiguration() into worker-specific and worker-independent
 * parts.
 *
 * TODO: should this be in the blob package?
 * @author <NAME> <<EMAIL>>
 * @since 7/27/2013
 */
public class BlobHostStreamCompiler implements StreamCompiler {
	private final BlobFactory blobFactory;
	public BlobHostStreamCompiler(BlobFactory blobFactory) {
		this.blobFactory = blobFactory;
	}

	@Override
	public <I, O> CompiledStream<I, O> compile(OneToOneElement<I, O> stream) {
		ConnectWorkersVisitor cwv = new ConnectWorkersVisitor();
		stream.visit(cwv);
		ImmutableSet<Worker<?, ?>> workers = Workers.getAllWorkersInGraph(cwv.getSource());
		Blob blob = blobFactory.makeBlob(workers, getConfiguration(workers), getMaxNumCores());

		Token inputToken = Iterables.getOnlyElement(blob.getInputs());
		Token outputToken = Iterables.getOnlyElement(blob.getOutputs());
		Buffer inputBuffer = Buffers.blockingQueueBuffer(new ArrayBlockingQueue<>(blob.getMinimumBufferCapacity(inputToken)), false, false);
		Buffer outputBuffer = Buffers.blockingQueueBuffer(new ArrayBlockingQueue<>(blob.getMinimumBufferCapacity(outputToken)), false, false);
		ImmutableMap<Token, Buffer> bufferMap = ImmutableMap.<Token, Buffer>builder()
				.put(inputToken, inputBuffer)
				.put(outputToken, outputBuffer)
				.build();
		blob.installBuffers(bufferMap);

		ImmutableList.Builder<PollingCoreThread> threads = ImmutableList.builder();
		for (int i = 0; i < blob.getCoreCount(); ++i) {
			PollingCoreThread thread = new PollingCoreThread(blob.getCoreCode(i), blob.toString()+"-"+i);
			threads.add(thread);
			thread.start();
		}

		return new BlobHostCompiledStream<>(blob, inputBuffer, outputBuffer, threads.build());
	}

	/**
	 * Get a configuration for the given worker.  This implementation returns
	 * the default configuration from the blob factory, but subclasses can
	 * override this to implement compiler-specific options.
	 * @param workers the set of workers in the blob
	 * @return a configuration
	 */
	protected Configuration getConfiguration(Set<Worker<?, ?>> workers) {
		return blobFactory.getDefaultConfiguration(workers);
	}

	protected int getMaxNumCores() {
		return 1;
	}

	private static final class BlobHostCompiledStream<I, O> implements CompiledStream<I, O> {
		private final Blob blob;
		private final Buffer inputBuffer, outputBuffer;
		private final ImmutableList<PollingCoreThread> threads;
		private BlobHostCompiledStream(Blob blob, Buffer inputBuffer, Buffer outputBuffer, ImmutableList<PollingCoreThread> threads) {
			this.blob = blob;
			this.inputBuffer = inputBuffer;
			this.outputBuffer = outputBuffer;
			this.threads = threads;
		}

		@Override
		public boolean offer(I input) {
			return inputBuffer.write(input);
		}

		@Override
		@SuppressWarnings("unchecked")
		public O poll() {
			return (O)outputBuffer.read();
		}
{'fraction_non_alphanumeric': 0.10163037010591455, 'fraction_numerical': 0.0020230869927406877, 'mean_word_length': 5.202214022140222, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 43, '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': '1461599', 'n_tokens_mistral': 2648, 'n_tokens_neox': 2465, 'n_words': 669}