<commit_before>
package cz.seznam.euphoria.core.executor;

import cz.seznam.euphoria.core.client.operator.Pair;
import cz.seznam.euphoria.core.client.dataset.Dataset;
import cz.seznam.euphoria.core.client.dataset.GroupedDataset;
import cz.seznam.euphoria.core.client.dataset.Partitioner;
import cz.seznam.euphoria.core.client.dataset.Partitioning;
import cz.seznam.euphoria.core.client.dataset.Window;
import cz.seznam.euphoria.core.client.dataset.Windowing;
import cz.seznam.euphoria.core.client.flow.Flow;
import cz.seznam.euphoria.core.client.functional.BinaryFunction;
import cz.seznam.euphoria.core.client.functional.UnaryFunction;
import cz.seznam.euphoria.core.client.functional.UnaryFunctor;
import cz.seznam.euphoria.core.client.graph.DAG;
import cz.seznam.euphoria.core.client.io.Collector;
import cz.seznam.euphoria.core.client.io.DataSink;
import cz.seznam.euphoria.core.client.io.DataSource;
import cz.seznam.euphoria.core.client.io.Partition;
import cz.seznam.euphoria.core.client.io.Writer;
import cz.seznam.euphoria.core.client.operator.FlatMap;
import cz.seznam.euphoria.core.client.operator.Operator;
import cz.seznam.euphoria.core.client.operator.ReduceStateByKey;
import cz.seznam.euphoria.core.client.operator.Repartition;
import cz.seznam.euphoria.core.client.operator.State;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Inmem executor for testing.
 */
public class InMemExecutor implements Executor {


  @FunctionalInterface
  private interface Supplier<T> {
    T get() throws EndOfStreamException;
  }

  // end of stream signal
  private static class EndOfStream {
    private static EndOfStream get() {
      return new EndOfStream();
    }
  }

  private static class EndOfStreamException extends Exception {
    
  }

  
  private static final class PartitionSupplierStream<T> implements Supplier<T> {
    final Iterator<T> it;
    PartitionSupplierStream(Partition<T> partition) {
      this.it = partition.iterator();
    }
    @Override
    public T get() throws EndOfStreamException {
      T next = it.next();
      if (next == null) {
        throw new EndOfStreamException();
      }
      return next;
    }
  }


  private static final class QueueCollector<T> implements Collector<T> {
    static <T> QueueCollector<T> wrap(BlockingQueue<T> queue) {
      return new QueueCollector<>(queue);
    }
    private final BlockingQueue<T> queue;
    QueueCollector(BlockingQueue<T> queue) {
      this.queue = queue;
    }
    @Override
    public void collect(T elem) {
      try {
        queue.put(elem);
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      }
    }
  }

  private static final class QueueSupplier<T> implements Supplier<T> {

    static <T> QueueSupplier<T> wrap(BlockingQueue<T> queue) {
      return new QueueSupplier<>(queue);
    }

    private final BlockingQueue<T> queue;

    QueueSupplier(BlockingQueue<T> queue) {
      this.queue = queue;
    }

    @Override
    public T get() throws EndOfStreamException {
      try {
        T take = queue.take();
        if (take instanceof EndOfStream) {
          throw new EndOfStreamException();
        }
        return take;
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      }
    }

  }


  private static final class ExecutionContext {
    Map<Dataset<?>, List<Supplier<?>>> materializedDatasets
        = Collections.synchronizedMap(new HashMap<>());

    private boolean containsKey(Dataset<?> d) {
      return materializedDatasets.containsKey(d);
    }
    void add(Dataset<?> dataset, List<Supplier<?>> partitions) {
      if (containsKey(dataset)) {
        throw new IllegalArgumentException("Dataset "
            + dataset + " is already materialized!");
      }
      materializedDatasets.put(dataset, partitions);
    }
    List<Supplier<?>> get(Dataset<?> ds) {
      List<Supplier<?>> sup = materializedDatasets.get(ds);
      if (sup == null) {
        throw new IllegalArgumentException(
            "Do not have suppliers for given dataset (original producer "
            + ds.getProducer() + ")");
      }
      return sup;
    }
  }

  private final BlockingQueue<Runnable> queue = new LinkedTransferQueue<>();
  private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
      Runtime.getRuntime().availableProcessors(),
      10 * Runtime.getRuntime().availableProcessors(),
      60,
      TimeUnit.SECONDS,
      queue,
      new ThreadFactory() {
        ThreadFactory factory = Executors.defaultThreadFactory();
        @Override
        public Thread newThread(Runnable r) {
          Thread thread = factory.newThread(r);
          thread.setUncaughtExceptionHandler((Thread t, Throwable e) -> {
            e.printStackTrace(System.err);
          });<commit_msg>#! Fix dead lock when executing many tasks<commit_after>
package cz.seznam.euphoria.core.executor;

import cz.seznam.euphoria.core.client.dataset.Dataset;
import cz.seznam.euphoria.core.client.dataset.GroupedDataset;
import cz.seznam.euphoria.core.client.dataset.Partitioner;
import cz.seznam.euphoria.core.client.dataset.Partitioning;
import cz.seznam.euphoria.core.client.dataset.Window;
import cz.seznam.euphoria.core.client.dataset.Windowing;
import cz.seznam.euphoria.core.client.flow.Flow;
import cz.seznam.euphoria.core.client.functional.BinaryFunction;
import cz.seznam.euphoria.core.client.functional.UnaryFunction;
import cz.seznam.euphoria.core.client.functional.UnaryFunctor;
import cz.seznam.euphoria.core.client.graph.DAG;
import cz.seznam.euphoria.core.client.io.Collector;
import cz.seznam.euphoria.core.client.io.DataSink;
import cz.seznam.euphoria.core.client.io.DataSource;
import cz.seznam.euphoria.core.client.io.Partition;
import cz.seznam.euphoria.core.client.io.Writer;
import cz.seznam.euphoria.core.client.operator.FlatMap;
import cz.seznam.euphoria.core.client.operator.Operator;
import cz.seznam.euphoria.core.client.operator.Pair;
import cz.seznam.euphoria.core.client.operator.ReduceStateByKey;
import cz.seznam.euphoria.core.client.operator.Repartition;
import cz.seznam.euphoria.core.client.operator.State;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Inmem executor for testing.
 */
public class InMemExecutor implements Executor {


  @FunctionalInterface
  private interface Supplier<T> {
    T get() throws EndOfStreamException;
  }

  // end of stream signal
  private static class EndOfStream {
    private static EndOfStream get() {
      return new EndOfStream();
    }
  }

  private static class EndOfStreamException extends Exception {
    
  }

  
  private static final class PartitionSupplierStream<T> implements Supplier<T> {
    final Iterator<T> it;
    PartitionSupplierStream(Partition<T> partition) {
      this.it = partition.iterator();
    }
    @Override
    public T get() throws EndOfStreamException {
      T next = it.next();
      if (next == null) {
        throw new EndOfStreamException();
      }
      return next;
    }
  }


  private static final class QueueCollector<T> implements Collector<T> {
    static <T> QueueCollector<T> wrap(BlockingQueue<T> queue) {
      return new QueueCollector<>(queue);
    }
    private final BlockingQueue<T> queue;
    QueueCollector(BlockingQueue<T> queue) {
      this.queue = queue;
    }
    @Override
    public void collect(T elem) {
      try {
        queue.put(elem);
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      }
    }
  }

  private static final class QueueSupplier<T> implements Supplier<T> {

    static <T> QueueSupplier<T> wrap(BlockingQueue<T> queue) {
      return new QueueSupplier<>(queue);
    }

    private final BlockingQueue<T> queue;

    QueueSupplier(BlockingQueue<T> queue) {
      this.queue = queue;
    }

    @Override
    public T get() throws EndOfStreamException {
      try {
        T take = queue.take();
        if (take instanceof EndOfStream) {
          throw new EndOfStreamException();
        }
        return take;
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      }
    }

  }


  private static final class ExecutionContext {
    Map<Dataset<?>, List<Supplier<?>>> materializedDatasets
        = Collections.synchronizedMap(new HashMap<>());

    private boolean containsKey(Dataset<?> d) {
      return materializedDatasets.containsKey(d);
    }
    void add(Dataset<?> dataset, List<Supplier<?>> partitions) {
      if (containsKey(dataset)) {
        throw new IllegalArgumentException("Dataset "
            + dataset + " is already materialized!");
      }
      materializedDatasets.put(dataset, partitions);
    }
    List<Supplier<?>> get(Dataset<?> ds) {
      List<Supplier<?>> sup = materializedDatasets.get(ds);
      if (sup == null) {
        throw new IllegalArgumentException(
            "Do not have suppliers for given dataset (original producer "
            + ds.getProducer() + ")");
      }
      return sup;
    }
  }

  private final BlockingQueue<Runnable> queue = new SynchronousQueue<>(false);
  private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
      0, Integer.MAX_VALUE,
      60,
      TimeUnit.SECONDS,
      queue,
      new ThreadFactory() {
        ThreadFactory factory = Executors.defaultThreadFactory();
        @Override
        public Thread newThread(Runnable r) {
          Thread thread = factory.newThread(r);
          thread.setUncaughtExceptionHandler((Thread t, Throwable e) -> {
            e.printStackTrace(System.err);
          });
{'fraction_non_alphanumeric': 0.10740122426266, 'fraction_numerical': 0.0006492301984789464, 'mean_word_length': 3.485440931780366, 'pattern_counts': {'":': 0, '<': 77, '<?xml version=': 0, '>': 79, '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': '2564725', 'n_tokens_mistral': 3348, 'n_tokens_neox': 3330, 'n_words': 670}