Collect partial results from parallel streams

Question

In Java8, processing pairs of items in two parallel streams as below:

final List<Item> items = getItemList();
final int l = items.size();
List<String> results = Collections.synchronizedList(new ArrayList<String>());
IntStream.range(0, l - 1).parallel().forEach(
    i -> {
        Item item1 = items.get(i);
        int x1 = item1.x;
        IntStream.range(i + 1, l).parallel()
            .forEach(j -> {
                Item item2 = items.get(j);
                int x2 = item2.x;
                if (x1 + x2 < 200) return;
                // code that writes to ConcurrentHashMap defined near results
                if (x1 + x2 > 500) results.add(i + " " + j);
            });
    }
);

Each stream pair writes to ConcurrentHashMap, and depending on certain conditions it may terminate the stream execution by calling return; or it may write to a synchronized list.

I want to make streams return the results like return i + " " + j and collect those results into a list strings outside. It should be partial as returning nothing must be supported (in case when x1 + x2 < 200).

What would be the most time-efficient (fastest code) way to achieve that?

Please provide correct code which compiles. Your `results` is declared as array, but you are using it like list. Where's the `ConcurrentHashMap`? What exactly are you writing there? Also please provide a sample input and the desired output for it. Now it's a little bit unclear what are you trying to achieve. — Tagir Valeev, Aug 07 '15 at 04:25
You are only adding to `results` if `x1 + x2 > 500`. Why not use a Collector? Also ... as far as I know, you can't return from a foreach like that. — Chthonic Project, Aug 07 '15 at 15:02
In your particular example, doing things in parallel may not yield faster performance. Check out this question and its top answer: http://stackoverflow.com/questions/23170832/java-8s-streams-why-parallel-stream-is-slower — Chthonic Project, Aug 07 '15 at 15:04
Before caring about time efficiency, I think you should care about correctness. (1) If you need to terminate the stream upon a certain condition, you cannot parallelize, otherwise you cannot control the execution order and there might be pairs added to the `results` list that occur logically after the pair that triggered your stop condition `x1 + x2 < 200`. (2) A `return;` like that is definitively not the way to stop a stream execution. — Helder Pereira, Aug 08 '15 at 18:51

score 0 · Answer 1 · answered Aug 08 '15 at 21:56

In this answer, I will not address the time efficiency, because there are correctness problems that should be handled beforehand.

As I said in the comments, it is not possible to stop the stream execution after a certain condition if we parallelize the stream. Otherwise, there might be some pairs (i,j) that are already being executed that are numerically after a pair that triggered the stop condition x1 + x2 < 200. Another issue is the return; inside the lambda, all it will do is skip the second if for the j for which x1 + x2 < 200 holds, but the stream will continue with j+1.

There is no straightforward way to stop a stream in Java, but we can achieve that with allMatch, as we can expect that as soon as it finds a false value, it will short-circuit and return false right way.

So, this would be a correct version of your code:

IntStream.range(0, l - 1).allMatch(i -> {
    int x1 = items.get(i).x;
    return IntStream.range(i + 1, l).allMatch(j -> {
        int x2 = items.get(j).x;
        if (x1 + x2 < 200) {
            return false;
        } else {
            if (x1 + x2 > 500) results2.add(i + " " + j);
            return true;
        }
    });
});

For the following example, with the constructor Item(int x, int y):

final List<Item> items = Arrays.asList(
        new Item(200, 0),
        new Item(100, 0),
        new Item(500, 0),
        new Item(400, 0),
        new Item(1, 0));

The contents of results in my version is:

[0 2, 0 3, 1 2]

With your code (order and elements vary in each execution):

[2 4, 2 3, 1 2, 0 3, 0 2]

score -1 · Answer 2 · answered Aug 07 '15 at 15:26

I think this will be more efficient (haven't done any micro benchmarking though):

IntStream.range(0,l-1).forEach(
    i -> IntStream.range(i+1,l)
                  .filter(j -> items.get(i).x + items.get(j).x > 500)
                  .forEach(j -> results.add(i + " " + j)));

However, if I was really worried about the time taken to do this, I'd pay more attention to what kind of a List implementation is used for items. Perhaps even convert the list to a HashMap<Integer, Item> before getting into the lambda. For example, if items is a LinkedList, any improvement to the lambda may be inconsequential because items.get() will eat up all the time.

@downvoter: an explanation and/or suggestions for improvement would be helpful, and also along the lines of SO decorum, as suggested here: http://meta.stackexchange.com/questions/135/encouraging-people-to-explain-downvotes — Chthonic Project, Aug 10 '15 at 19:00

Collect partial results from parallel streams

2 Answers2