<commit_before>package com.azure.cosmos.dotnet.benchmark;

import com.azure.cosmos.implementation.guava25.base.Function;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.DoubleStream;

public class ParallelExecutionStrategy implements IExecutionStrategy {
    private static final long OUTPUT_LOOP_DELAY_IN_MS = 1000;
    private final Function<Void, IBenchmarkOperation> benchmarkOperation;
    private final BenchmarkConfig config;
    private final AtomicInteger pendingExecutorCount = new AtomicInteger();

    public ParallelExecutionStrategy(BenchmarkConfig config,
                                     Function<Void, IBenchmarkOperation> benchmarkOperation) {

        this.config = config;
        this.benchmarkOperation = benchmarkOperation;
    }

    @Override
    public RunSummary execute(
        int serialExecutorConcurrency,
        int serialExecutorIterationCount,
        boolean traceFailures,
        double warmupFraction) {

        IExecutor warmupExecutor = new SerialOperationExecutor(
            "Warmup",
            this.benchmarkOperation.apply(null));

        // Block while warmup happens
        warmupExecutor.execute(
            (int)(serialExecutorIterationCount * warmupFraction),
            true,
            traceFailures,
            null).block();

        this.pendingExecutorCount.set(serialExecutorConcurrency);

        // Setting up executors and kick-off tests
        final IExecutor[] executors = new IExecutor[serialExecutorConcurrency];

        ArrayList<Double> perLoopCounters = new ArrayList<>();
        AtomicReference<Summary> lastSummary = new AtomicReference<>();
        lastSummary.set(new Summary());
        final AtomicLong startTime = new AtomicLong();

        Flux<OperationResult> processingFlux = Flux
            .range(0, serialExecutorConcurrency)
            .subscribeOn(Schedulers.elastic())
            .flatMap((i) -> {
                    executors[i] = new SerialOperationExecutor(
                        String.valueOf(i),
                        this.benchmarkOperation.apply(null));

                    return executors[i].execute(
                        serialExecutorIterationCount,
                        false,
                        traceFailures,
                        this.pendingExecutorCount::decrementAndGet);
                },
                serialExecutorConcurrency)
            .doFirst(() -> {
                Utility.traceInformation("Starting executors...");
                startTime.set(System.nanoTime());
            });

        Flux<Boolean> monitoringTimer = Flux
            .interval(Duration.ofMillis(OUTPUT_LOOP_DELAY_IN_MS))
            .onBackpressureDrop()
            .map((time) -> {
                int pendingExecutorCountSnapshot = this.pendingExecutorCount.get();
                boolean isLastIterationCompleted = pendingExecutorCountSnapshot <= 0;

                Summary currentTotalSummary = new Summary();
                for (IExecutor executor : executors) {
                    Summary executorSummary = new Summary(
                        executor.getSuccessOperationCount(),
                        executor.getFailedOperationCount(),
                        executor.getTotalRuCharges());

                    currentTotalSummary = currentTotalSummary.add(executorSummary);
                }

                // In-theory summary might be lower than real as its not transactional on time
                currentTotalSummary.setElapsedTimeInMs(
                    TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTime.get(),
                        TimeUnit.NANOSECONDS));

                Summary lastSummarySnapshot = lastSummary.getAndSet(currentTotalSummary);
                Summary diff = currentTotalSummary.subtract(lastSummarySnapshot);

                diff.print(currentTotalSummary.getTotalOperationsCount());<commit_msg>Internal: Fixed NPE in dotnet benchmark tool when concurrency is very high (#15645)<commit_after>package com.azure.cosmos.dotnet.benchmark;

import com.azure.cosmos.implementation.guava25.base.Function;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.DoubleStream;

public class ParallelExecutionStrategy implements IExecutionStrategy {
    private static final long OUTPUT_LOOP_DELAY_IN_MS = 1000;
    private final Function<Void, IBenchmarkOperation> benchmarkOperation;
    private final BenchmarkConfig config;
    private final AtomicInteger pendingExecutorCount = new AtomicInteger();

    public ParallelExecutionStrategy(BenchmarkConfig config,
                                     Function<Void, IBenchmarkOperation> benchmarkOperation) {

        this.config = config;
        this.benchmarkOperation = benchmarkOperation;
    }

    @Override
    public RunSummary execute(
        int serialExecutorConcurrency,
        int serialExecutorIterationCount,
        boolean traceFailures,
        double warmupFraction) {

        IExecutor warmupExecutor = new SerialOperationExecutor(
            "Warmup",
            this.benchmarkOperation.apply(null));

        // Block while warmup happens
        warmupExecutor.execute(
            (int)(serialExecutorIterationCount * warmupFraction),
            true,
            traceFailures,
            null).block();

        this.pendingExecutorCount.set(serialExecutorConcurrency);

        // Setting up executors and kick-off tests
        final IExecutor[] executors = new IExecutor[serialExecutorConcurrency];

        ArrayList<Double> perLoopCounters = new ArrayList<>();
        AtomicReference<Summary> lastSummary = new AtomicReference<>();
        lastSummary.set(new Summary());
        final AtomicLong startTime = new AtomicLong();

        Flux<OperationResult> processingFlux = Flux
            .range(0, serialExecutorConcurrency)
            .subscribeOn(Schedulers.elastic())
            .flatMap((i) -> {
                    executors[i] = new SerialOperationExecutor(
                        String.valueOf(i),
                        this.benchmarkOperation.apply(null));

                    return executors[i].execute(
                        serialExecutorIterationCount,
                        false,
                        traceFailures,
                        this.pendingExecutorCount::decrementAndGet);
                },
                serialExecutorConcurrency)
            .doFirst(() -> {
                Utility.traceInformation("Starting executors...");
                startTime.set(System.nanoTime());
            });

        Flux<Boolean> monitoringTimer = Flux
            .interval(Duration.ofMillis(OUTPUT_LOOP_DELAY_IN_MS))
            .onBackpressureDrop()
            .map((time) -> {
                int pendingExecutorCountSnapshot = this.pendingExecutorCount.get();
                boolean isLastIterationCompleted = pendingExecutorCountSnapshot <= 0;

                Summary currentTotalSummary = new Summary();
                for (IExecutor executor : executors) {
                    if (executor == null) {
                        continue;
                    }

                    Summary executorSummary = new Summary(
                        executor.getSuccessOperationCount(),
                        executor.getFailedOperationCount(),
                        executor.getTotalRuCharges());

                    currentTotalSummary = currentTotalSummary.add(executorSummary);
                }

                // In-theory summary might be lower than real as its not transactional on time
                currentTotalSummary.setElapsedTimeInMs(
                    TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTime.get(),
                        TimeUnit.NANOSECONDS));

                Summary lastSummarySnapshot = lastSummary.getAndSet(currentTotalSummary);
                Summary diff = currentTotalSummary.subtract(lastSummarySnapshot);

                diff.print(currentTotalSummary.getTotalOperationsCount());
{'fraction_non_alphanumeric': 0.07766433405988299, 'fraction_numerical': 0.0024090856946197086, 'mean_word_length': 2.2445106066244884, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3529701', 'n_tokens_mistral': 2139, 'n_tokens_neox': 1995, 'n_words': 407}