#!/usr/bin/env python3
"""Demonstrate Parallel Python callbacks.

Website: https://www.parallelpython.com
License: Apache-2.0 (see LICENSE and NOTICE)

Calculates the partial sum 1 - 1/2 + 1/3 - 1/4 + 1/5 - 1/6 + ...
(in the limit it converges to ln(2)).

Usage:
    python callback.py [ncpus]
"""

import math
import sys
import threading

import pp


class Sum:
    """Thread-safe accumulator for callback results."""

    def __init__(self):
        self.value = 0.0
        self.lock = threading.Lock()

    def add(self, value):
        """The callback function."""
        with self.lock:
            self.value += value


def part_sum(start, end):
    """Calculate a partial alternating harmonic sum."""
    s = 0.0
    for x in range(start, end):
        if x % 2 == 0:
            s -= 1.0 / x
        else:
            s += 1.0 / x
    return s


def main():
    print("""Usage: python callback.py [ncpus]
    [ncpus] - the number of workers to run in parallel,
    if omitted it will be set to the number of processors in the system""")

    start = 1
    end = 20000000

    # Divide the task into 128 subtasks
    parts = 128
    step = (end - start) // parts + 1

    # tuple of all parallel python servers to connect with
    ppservers = ()

    if len(sys.argv) > 1:
        ncpus = int(sys.argv[1])
        job_server = pp.Server(ncpus, ppservers=ppservers)
    else:
        job_server = pp.Server(ppservers=ppservers)

    print(f"Starting pp with {job_server.get_ncpus()} workers")

    # Create an instance of callback class
    summation = Sum()

    # Execute the same task with different data
    import time
    start_time = time.time()
    for index in range(parts):
        starti = start + index * step
        endi = min(start + (index + 1) * step, end)
        # Submit a job which will calculate partial sum
        job_server.submit(part_sum, (starti, endi), callback=summation.add)

    # Wait for jobs in all groups to finish
    job_server.wait()

    # Print the partial sum
    print(f"Partial sum is {summation.value} | diff = {math.log(2) - summation.value}")

    print(f"Time elapsed: {time.time() - start_time:.3f}s")
    job_server.print_stats()
    job_server.destroy()


if __name__ == "__main__":
    main()