Using streams to compute infinite sum

Question

I'm trying to learn the details of using the Stream API, and one of the assignments I gave myself was to try to write a method that takes an infinite DoubleStream and tries to compute the sum (assuming it converges). That is, I'd like to write a method

public static double infiniteSum(DoubleStream ds) { ... }

that I could call with something like

double sum = infiniteSum(IntStream.iterate(1, (i -> i + 1))
                                  .mapToDouble(n -> 1 / ((double)n * n)));

to get the sum (1 + 1/2² + 1/3² + ... ) = ζ(2) = π²/6.

My crude method for doing this the old way:

public static void yeOldeWaye() {
    double sum = 0;
    for (double n = 1; ; n++) {
        double term = 1 / (n * n);
        if (Math.abs(term) <= 1e-12 * Math.abs(sum)) {
            break;
        }
        sum += term;
    }
    System.out.println(sum);
}

which gives me a result accurate to 5 places.

I can implement the method in a hacked way using iterator():

public static double infiniteSum1(DoubleStream ds) {
    double sum = 0;
    PrimitiveIterator.OfDouble it = ds.iterator();
    while (true) {
        double term = it.next();
        if (Math.abs(term) <= 1e-12 * Math.abs(sum)) {
            break;
        }
        sum += term;
    }
    return sum;
}

but that just feels like reverting to the old way, and I was looking for a method that used streams more the way they were intended to be used, or something.

This produces the correct result:

private static class DoubleAccumulator {
    public double sum;
    public DoubleAccumulator() {
        sum = 0;
    }
}

public static double infiniteSum(DoubleStream ds) {
    DoubleAccumulator summer = ds.limit(800000).collect
        (DoubleAccumulator::new,
         (s, d) -> s.sum += d,
         (s1, s2) -> s1.sum += s2.sum);
    return summer.sum;
}

but I happened to know that the old method used almost 800000 terms, and putting a limit on the stream defeats my purpose. The problem is that I don't see a way to cut off a stream other than by using limit(), which means that I have to know beforehand how many terms I'm going to have; I don't see a way to stop the stream based on some condition that's computed based on what I'm seeing in the stream.

This doesn't work:

public static double infiniteSum(DoubleStream ds) {
    DoubleAccumulator summer = ds.collect
        (DoubleAccumulator::new,
         (s, d) -> { if (Math.abs(d) <= 1e-12 * Math.abs(s.sum)) {
                        ds.close();  // AAACK
                     } else
                        s.sum += d;
                   },
         (s1, s2) -> s1.sum += s2.sum);
    return summer.sum;
}

A trace indicates that something does happen when the last term is seen, but nothing good: in one case, the computation stopped but the program still hung, and in another case, it gave me a cute little crash dump that I get to report to Oracle.

So is there a way to accomplish the sort of thing I'm looking for?

(Note: I'm assuming serial streams, for now. But I think this is the sort of problem that could benefit from parallelism, once I figure out how to make it work.)

Answer 1

If you have such a dependency between the termination criteria and the Collector’s result, using mutable state is unavoidable. But as long as you don’t need parallel execution, the implementation can be straight-forward:

class MutableDouble {
    double sum;
}
MutableDouble sumHolder=new MutableDouble();
DoubleStream ds=IntStream.iterate(1, (i -> i + 1))
                         .mapToDouble(n -> 1 / ((double)n * n));
ds.peek(term -> sumHolder.sum+=term)
  .filter(term -> Math.abs(term)<1e-12*Math.abs(sumHolder.sum))
  .findFirst();
System.out.println(sumHolder.sum);

Since it will sum up before comparing it is not exactly the same as your original logic but rather like:

double sum = 0;
for (double n = 1; ; n++) {
    double term = 1 / (n * n);
    sum += term;
    if (Math.abs(term) <= 1e-12 * Math.abs(sum)) {
        break;
    }
}

If you insist on the original logic it has to be slightly more complicated:

class MutableDouble {
    double sum, next;
    void add(double d) {
        sum=next;
        next+=d;
    }
}
MutableDouble sumHolder=new MutableDouble();
DoubleStream ds=IntStream.iterate(1, (i -> i + 1))
                         .mapToDouble(n -> 1 / ((double)n * n));
ds.peek(sumHolder::add)
  .filter(term -> Math.abs(term)<1e-12*Math.abs(sumHolder.sum))
  .findFirst();

Answer 2

Let's assume that you know that the result is ~ 1. So your comparison: term <= 1e-12 * sum is roughly equivalent to term <= 1e-12.

Now you don't need the sum at each step which simplifies the problem a bit (otherwise you can keep track of the sum in the iterator). It can then be written:

public static void yeStreamsWaye() {
    System.out.println(stream().sum());
}

private static DoubleStream stream() {
    return StreamSupport.doubleStream(Spliterators.spliteratorUnknownSize(new Piterator(),
            Spliterator.ORDERED | Spliterator.IMMUTABLE | Spliterator.NONNULL), false);
}

static class Piterator implements PrimitiveIterator.OfDouble {
    private double t = 1;
    @Override public boolean hasNext() { return 1 / (t * t) > 1e-12; }
    @Override public double nextDouble() { return 1 / (t * t++); }
}

It doesn't look like this can be easily "parallelised" and I would tend to conclude that your initial loop doesn't look so bad...