package net.ranides.assira.collection.query.support;

import lombok.experimental.UtilityClass;

import net.ranides.assira.functional.Consumers;
import net.ranides.assira.functional.Predicates;
import net.ranides.assira.math.MathUtils;

import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.List;
import java.util.Spliterator;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.StreamSupport;

/**
 * This is internal utility class not supposed to use by clients.
 * It has default implementations of parallel query operations using only Spliterator contract.
 *
 * Those methods should not be called directly, but by "BaseSpliterator" facade.
 */
@UtilityClass
@SuppressWarnings("MissingJavadoc")
public class BaseSpliteratorParalell {

    /**
     * This number tells how many additional splits are introduced into every working task.
     * Value 3 means, that every fragment will be roughly 8 times shorter than "theoretical exact".
     */
    private static final int SPLIT_FACTOR = 3;

    /**
     * This number is the minimum size of working task. Tasks with smaller size won't be split.
     */
    private static final int SPLIT_BOUND = 3;

    public static <T> void forEach(Spliterator<T> that, Consumer<? super T> consumer) {
        StreamSupport.stream(that, true).forEach(consumer);
    }

    public static <T> void forEach(Spliterator<T> that, Consumers.EachConsumer<? super T> consumer) {
        if(!that.hasCharacteristics(Spliterator.SUBSIZED)) {
            BaseState.Index index = BaseState.newIndex(true);
            StreamSupport.stream(that, true).forEach(v -> consumer.accept(index.next(), v));
            return;
        }

        ForkJoinPool executor = ForkJoinPool.commonPool();
        int splits = estimateSplits(executor, that);
        List<Callable<Void>> tasks = new ArrayList<>();
        try {
            forEachSplit(splits, tasks, that, 0, consumer);
            executor.invokeAll(tasks);
        } finally {
            executor.shutdown();
        }
    }

    private static <T> void forEachSplit(int splits, List<Callable<Void>> tasks, Spliterator<T> that, int offset, Consumers.EachConsumer<? super T> consumer) {
        Spliterator<T> prefix = null;
        if(splits > 0) {
            prefix = that.trySplit();
        }

        if(prefix != null) {
            int delta = (int) prefix.getExactSizeIfKnown();
            forEachSplit(splits-1, tasks, prefix, offset, consumer);
            forEachSplit(splits-1, tasks, that, offset + delta, consumer);
        } else {
            tasks.add(() -> {
                BaseState.Index index = BaseState.newIndex(false, offset);
                that.forEachRemaining(v -> consumer.accept(index.next(), v));
                return null;
            });
        }
    }

    public static <T> boolean whileEach(Spliterator<T> that, Predicate<? super T> consumer) {
        return StreamSupport.stream(that, true).allMatch(consumer);
    }

    public static <T> boolean whileEach(Spliterator<T> that, Predicates.EachPredicate<? super T> consumer) {
        if(!that.hasCharacteristics(Spliterator.SUBSIZED)) {
            BaseState.Index index = BaseState.newIndex(true);
            return StreamSupport.stream(that, true).allMatch(v -> consumer.test(index.next(), v));
        }

        ForkJoinPool executor = ForkJoinPool.commonPool();
        int splits = estimateSplits(executor, that);
        List<Callable<Void>> tasks = new ArrayList<>();
        BaseState.Terminator terminator = BaseState.newTerminator(true);
        try {
            whileEachSplit(splits, tasks, terminator, that, 0, consumer);
            executor.invokeAll(tasks);
            return !terminator.stopped();
        } finally {
            executor.shutdown();
        }
    }

    private static <T> void whileEachSplit(int splits, List<Callable<Void>> tasks, BaseState.Terminator terminator, Spliterator<T> that, int offset, Predicates.EachPredicate<? super T> consumer) {
        Spliterator<T> prefix = null;
        if(splits > 0) {
            prefix = that.trySplit();
        }

        if(prefix != null) {
            int delta = (int) prefix.getExactSizeIfKnown();
            whileEachSplit(splits-1, tasks, terminator, prefix, offset, consumer);
            whileEachSplit(splits-1, tasks, terminator, that, offset + delta, consumer);
        } else {
            tasks.add(() -> {
                BaseState.Index index = BaseState.newIndex(true, offset);
                StreamSupport.stream(that, false).allMatch(v -> {
                    if(terminator.stopped()) {
                        return false;
                    }
                    boolean test = consumer.test(index.next(), v);
                    if(!test) {
                        terminator.stop();
                    }
                    return test;
                });
                return null;
            });
        }
    }

    public static <T> int size(Spliterator<T> that) {
        if(that.hasCharacteristics(Spliterator.SIZED)) {
            return (int) that.estimateSize();
        }

        ForkJoinPool executor = ForkJoinPool.commonPool();
        int splits = estimateSplits(executor, that);
        List<Callable<Integer>> tasks = new ArrayList<>();
        try {
            sizeSplit(splits, tasks, that);
            int size = 0;
            for (Future<Integer> future : executor.invokeAll(tasks)) {
                size += future.get();
            }
            return size;
        } catch (ExecutionException | InterruptedException e) {
            throw new ConcurrentModificationException(e);
        } finally {
            executor.shutdown();
        }
    }

    private static <T> void sizeSplit(int splits, List<Callable<Integer>> tasks, Spliterator<T> that) {
        Spliterator<T> prefix = null;
        if(splits > 0) {
            prefix = that.trySplit();
        }

        if(prefix != null) {
            sizeSplit(splits-1, tasks, prefix);
            sizeSplit(splits-1, tasks, that);
        } else {
            tasks.add(() -> BaseSpliteratorSequential.size(that));
        }
    }

    private static int estimateSplits(ForkJoinPool executor, Spliterator<?> that) {
        // we can't split spliterators into groups with specified size
        // which mean we can't distribute elements evenly
        // we can do 2 things:
        //  - split into number of tasks lower than executor
        //  - split into number of tasks SIGNIFICANTLY bigger than executor

        // this estimation assumes that 3 additional splits will make groups
        // 8 times smaller, which will eliminate disproportion between groups
        //
        // effectively, it will introduce 1/8 overhead at most, which means ~12%

        int splits = MathUtils.log2floor(executor.getParallelism());

        long n = that.getExactSizeIfKnown();
        for(int i = 0; i< SPLIT_FACTOR; i++, splits++) {
            if(n / (1L << splits) <= SPLIT_BOUND) {
                break;
            }
        }
        return splits;
    }


}
