parallel-programming/src/main/java/ru/lionarius/IntegralCalculator.java

package ru.lionarius;

import ru.lionarius.sync.MyCountDownLatch;
import ru.lionarius.sync.MyReentrantLock;
import ru.lionarius.sync.MySemaphore;

import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Function;

/**
 * A class that provides a numerical method to calculate definite integrals of a given function.
 * The integration is performed using the trapezoidal rule, with a specified accuracy.
 */
public class IntegralCalculator {

    /**
     * The accuracy level for the integration process.
     * A smaller value results in higher accuracy but more computation time.
     */
    private final double accuracy;

    /**
     * A callback to track the progress of the integration.
     * The callback is called at least twice: once at the beginning, and once at the end of computation.
     * Takes the current step and the total number of steps.
     */
    private final BiConsumer<Long, Long> progressCallback;

    /**
     * Constructs an {@link IntegralCalculator} with the specified accuracy and an optional progress callback.
     * The progress callback is invoked at each step of the integration process.
     *
     * @param accuracy         The desired accuracy for the integral calculation.
     * @param progressCallback A callback function to track progress. Can be null.
     * @throws IllegalArgumentException if the accuracy is less than or equal to zero.
     */
    public IntegralCalculator(double accuracy, final BiConsumer<Long, Long> progressCallback) {
        if (accuracy <= 0.0)
            throw new IllegalArgumentException("accuracy must be a positive number");

        this.accuracy = accuracy;
        this.progressCallback = progressCallback;
    }

    /**
     * Constructs an {@link IntegralCalculator} with the specified accuracy.
     *
     * @param accuracy The desired accuracy for the integral calculation
     * @throws IllegalArgumentException if the accuracy is less than or equal to zero.
     */
    public IntegralCalculator(double accuracy) {
        this(accuracy, null);
    }

    /**
     * Returns the accuracy level used for the integration calculation.
     *
     * @return The accuracy level as a double.
     */
    public double getAccuracy() {
        return this.accuracy;
    }

    /**
     * Calculates the definite integral of the specified function over a given interval
     * using the trapezoidal rule.
     *
     * @param function The function to integrate.
     * @param lower    The lower bound of the integration interval.
     * @param upper    The upper bound of the integration interval.
     * @return The estimated value of the definite integral.
     * @throws NullPointerException if the function is null.
     */
    public double calculate(final Function<Double, Double> function, double lower, double upper, ExecutorService executor, int parallelism) throws ExecutionException, InterruptedException {
        if (function == null)
            throw new NullPointerException("function cannot be null");

        if (executor == null)
            throw new NullPointerException("executor cannot be null");

        if (parallelism <= 0)
            throw new IllegalArgumentException("parallelism must be a positive number");

        if (lower == upper) {
            this.callProgressCallback(0, 1);
            this.callProgressCallback(1, 1);

            return 0.0;
        }

        var invert = false;
        if (lower > upper) {
            invert = true;

            var temp = lower;
            lower = upper;
            upper = temp;
        }

        final var lowerBound = lower;
        final var upperBound = upper;

        var n = (long) Math.ceil((upperBound - lowerBound) / this.accuracy);
        var h = (upperBound - lowerBound) / n;

        this.callProgressCallback(0, n);

        var remainder = n % parallelism;
        var partitionSize = n / parallelism;
        var currentStart = 0L;

        var sum = (function.apply(lowerBound) + function.apply(upperBound)) / 2;

        final var futures = new ArrayList<Future<Double>>();
        final var semaphore = new MySemaphore(2);
        final var lock = new MyReentrantLock();
        final var latch = new MyCountDownLatch(parallelism);

        var times = new long[parallelism];
        var totalProgress = new long[]{0L};
        for (var i = 0; i < parallelism; i++) {
            final var start = currentStart;
            final var end = currentStart + partitionSize + (remainder > 0 ? 1 : 0);
            if (remainder > 0)
                remainder--;
            currentStart = end;

            final var innerI = i;

            futures.add(executor.submit(() -> {
                var partitionSum = 0.0;
                try {
                    semaphore.acquire();

                    for (var j = start; j < end; j++) {
                        partitionSum += function.apply(lowerBound + h * j);

                        lock.lock();
                        try {
                            totalProgress[0] += 1;
                            this.callProgressCallback(totalProgress[0], n);
                        } finally {
                            lock.unlock();
                        }
                    }

                    lock.lock();
                    try {
                        times[innerI] = System.nanoTime();
                    } finally {
                        lock.unlock();
                    }
                } finally {
                    semaphore.release();
                }
                
                latch.countDown();

                return partitionSum;
            }));
        }

        latch.await();

        for (var future : futures) {
            sum += future.resultNow();
        }

        var minimumTime = Long.MAX_VALUE;
        for (var time : times) {
            if (time < minimumTime)
                minimumTime = time;
        }

        for (var i = 0; i < parallelism; i++) {
            System.out.printf("Thread %d took %.6fms%n", i, (times[i] - minimumTime) / 1_000_000.0);
        }

        sum *= h;

        if (invert)
            sum = -sum;

        this.callProgressCallback(n, n);

        return sum;
    }

    /**
     * Calls the progress callback, if provided.
     *
     * @param current The current step in the integration process.
     * @param total   The total number of steps in the integration process.
     */
    private void callProgressCallback(long current, long total) {
        if (this.progressCallback != null)
            this.progressCallback.accept(current, total);
    }
}