<commit_before>import {ParallelAction} from "./parallel-action";
import {TaskDefinition} from "../task/task-definition";
import {IParallelGenerator} from "./parallel-generator";
import {ParallelWorkerFunctions} from "./parallel-worker-functions";
import {DefaultInitializedParallelOptions} from "./parallel-options";
import {IParallelStream, ParallelStreamImpl} from "./parallel-stream";

export interface IParallelTaskScheduling {
    numberOfWorkers: number;
    valuesPerWorker: number;
}

export interface IParallelChain<TIn, TOut> {
    map<TResult>(mapper: (this: void, element: TOut, index: number) => TResult): IParallelChain<TIn, TResult>;

    reduce(defaultValue: TOut, accumulator: (this: void, memo: TOut, value: TOut, index: number) => TOut): Promise<TOut>;
    reduce<TResult>(defaultValue: TResult, accumulator: (this: void, memo: TResult, value: TOut, index: number) => TResult, combiner: (this: void, subResult1: TResult, subResult2: TResult) => TResult): Promise<TResult>;

    filter(predicate: (this: void, value: TOut, index: number) => boolean): IParallelChain<TIn, TOut>;

    // sortBy?

    result(): IParallelStream<TOut[], TOut[]>;
}

export function toParallelChain<T>(generator: IParallelGenerator, options: DefaultInitializedParallelOptions, actions: ParallelAction[] = []): IParallelChain<T, T> {
    return new ParallelChainImpl<T, T>(generator, actions, options);
}

export class ParallelChainImpl<TIn, TOut> implements IParallelChain<TIn, TOut> {

    constructor(public generator: IParallelGenerator, private actions: ParallelAction[] = [], private options: DefaultInitializedParallelOptions) {
    }

    public map<TResult>(mapper: (this: void, element: TOut) => TResult): IParallelChain<TIn, TResult> {
        return this._chain<TResult>(ParallelWorkerFunctions.map, mapper);
    }

    public reduce<TResult>(defaultValue: TResult, accumulator: (this: void, memo: TResult, value: TOut) => TResult, combiner?: (this: void, sub1: TResult, sub2: TResult) => TResult): Promise<TResult> {
        const stream = this._chain(ParallelWorkerFunctions.reduce, accumulator, defaultValue)._stream((intermediateResults: TResult[][]) => {
            let sum = defaultValue;
            let combineOperation: (accumulatedValue: TResult, value: TResult) => TResult = combiner || accumulator as any;

            for (const temp of intermediateResults) {
                sum = combineOperation(sum, temp[0]);
            }

            return sum;
        });
        return Promise.resolve(stream);
    }

    public filter(predicate: (this: void, value: TOut) => boolean): IParallelChain<TIn, TOut> {
        return this._chain<TOut>(ParallelWorkerFunctions.filter, predicate);
    }

    public result(): IParallelStream<TOut[], TOut[]> {
        return this._stream<TOut[], TOut[]>((intermediateValue: TOut[][]) => {
            if (intermediateValue.length === 0) {
                return [];
            }
            const [head, ...tail] = intermediateValue;
            return Array.prototype.concat.apply(head, tail);
        });
    }

    public getParallelTaskScheduling(totalItems: number): IParallelTaskScheduling {
        let itemsPerWorker = totalItems / this.options.maxConcurrencyLevel;

        if (this.options.minValuesPerWorker) {
            itemsPerWorker = Math.min(Math.max(itemsPerWorker, this.options.minValuesPerWorker), totalItems);
        }

        if (this.options.maxValuesPerWorker) {
            itemsPerWorker = Math.min(itemsPerWorker, this.options.maxValuesPerWorker);
        }<commit_msg>change api of reduce to return a parallel stream instead of a promise<commit_after>import {ParallelAction} from "./parallel-action";
import {TaskDefinition} from "../task/task-definition";
import {IParallelGenerator} from "./parallel-generator";
import {ParallelWorkerFunctions} from "./parallel-worker-functions";
import {DefaultInitializedParallelOptions} from "./parallel-options";
import {IParallelStream, ParallelStreamImpl} from "./parallel-stream";

export interface IParallelTaskScheduling {
    numberOfWorkers: number;
    valuesPerWorker: number;
}

export interface IParallelChain<TIn, TOut> {
    map<TResult>(mapper: (this: void, element: TOut, index: number) => TResult): IParallelChain<TIn, TResult>;

    reduce(defaultValue: TOut, accumulator: (this: void, memo: TOut, value: TOut, index: number) => TOut): IParallelStream<TOut[], TOut>;
    reduce<TResult>(defaultValue: TResult, accumulator: (this: void, memo: TResult, value: TOut, index: number) => TResult, combiner: (this: void, subResult1: TResult, subResult2: TResult) => TResult): IParallelStream<TResult[], TResult>;

    filter(predicate: (this: void, value: TOut, index: number) => boolean): IParallelChain<TIn, TOut>;

    // sortBy?

    result(): IParallelStream<TOut[], TOut[]>;
}

export function toParallelChain<T>(generator: IParallelGenerator, options: DefaultInitializedParallelOptions, actions: ParallelAction[] = []): IParallelChain<T, T> {
    return new ParallelChainImpl<T, T>(generator, actions, options);
}

export class ParallelChainImpl<TIn, TOut> implements IParallelChain<TIn, TOut> {

    constructor(public generator: IParallelGenerator, private actions: ParallelAction[] = [], private options: DefaultInitializedParallelOptions) {
    }

    public map<TResult>(mapper: (this: void, element: TOut) => TResult): IParallelChain<TIn, TResult> {
        return this._chain<TResult>(ParallelWorkerFunctions.map, mapper);
    }

    public reduce<TResult>(defaultValue: TResult, accumulator: (this: void, memo: TResult, value: TOut) => TResult, combiner?: (this: void, sub1: TResult, sub2: TResult) => TResult): IParallelStream<TResult[], TResult> {
        const stream = this._chain(ParallelWorkerFunctions.reduce, accumulator, defaultValue)._stream((intermediateResults: TResult[][]) => {
            let sum = defaultValue;
            let combineOperation: (accumulatedValue: TResult, value: TResult) => TResult = combiner || accumulator as any;

            for (const temp of intermediateResults) {
                sum = combineOperation(sum, temp[0]);
            }

            return sum;
        });
        return stream;
    }

    public filter(predicate: (this: void, value: TOut) => boolean): IParallelChain<TIn, TOut> {
        return this._chain<TOut>(ParallelWorkerFunctions.filter, predicate);
    }

    public result(): IParallelStream<TOut[], TOut[]> {
        return this._stream<TOut[], TOut[]>((intermediateValue: TOut[][]) => {
            if (intermediateValue.length === 0) {
                return [];
            }
            const [head, ...tail] = intermediateValue;
            return Array.prototype.concat.apply(head, tail);
        });
    }

    public getParallelTaskScheduling(totalItems: number): IParallelTaskScheduling {
        let itemsPerWorker = totalItems / this.options.maxConcurrencyLevel;

        if (this.options.minValuesPerWorker) {
            itemsPerWorker = Math.min(Math.max(itemsPerWorker, this.options.minValuesPerWorker), totalItems);
        }

        if (this.options.maxValuesPerWorker) {
            itemsPerWorker = Math.min(itemsPerWorker, this.options.maxValuesPerWorker);
        }
{'fraction_non_alphanumeric': 0.12706339298099598, 'fraction_numerical': 0.0016645859342488557, 'mean_word_length': 4.031402651779484, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 71, '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': '6799584', 'n_tokens_mistral': 2038, 'n_tokens_neox': 2056, 'n_words': 495}