fold throws NPE when using parallel() variant on the collection

[Paul Sandoz]

HI Arul,

It may be of benefit (apologies if you know this already) to point out that for your example "anyMatch" is likely to be more efficient than "fold", since the former need not consume all elements of the stream to produce a result.

However, while the current "anyMatch" implementation may return a result before consuming all elements, it currently does not cancel the creation of F/J tasks for other parts of the computation tree once a result is found, so the algorithm is not as efficient as it can be.

Paul.

On Oct 20, 2012, at 9:15 PM, Arul Dhesiaseelan <aruld at acm.org> wrote:

> Sorry for the noise. Thanks bitter_fox and Brian.
> 
> On Sat, Oct 20, 2012 at 12:09 AM, bitter_fox <bitterfoxc at gmail.com> wrote:
> 
>> Hi Arul,
>> the NPE was made by the third parameter.
>> 
>> The third parameter is used in parallel when fold reduces from Boolean and
>> Boolean to Boolean. So you have to provide it.
>> I think this will work well:
>> 
>> greetings.parallel().fold(() -> false, (Boolean b, String keyword) -> b ||
>> tweet.text.contains(keyword), (l, r) -> l | r)
>> 
>> Regards,
>> bitter_fox
>> 
>> I forgot to send to lambda-dev, sorry resending.
>> 
>> 
>> 2012/10/20 Arul Dhesiaseelan <aruld at acm.org>
>> 
>>> Hi,
>>> 
>>> This code used to work before, but it fails with build 61 and with the
>>> latest sources. It works fine with stream, but not in parallel in the fold
>>> context.
>>> 
>>>    List<String> greetings = Arrays.asList("hello", "Hello");
>>>    List<Tweet> tweets = new ArrayList<>();
>>>    tweets.add(new Tweet("arul", "Aloha!", 2));
>>>    tweets.add(new Tweet("joe", "Hello!", 16));
>>>    tweets.add(new Tweet("matt", "Hola!", 78));
>>> 
>>>    List<Tweet> filtered = tweets.parallel().filter(t ->
>>> greetings.parallel().anyMatch(t.text::contains)).into(new
>>> ArrayList<Tweet>());//greetings.parallel() works
>>>    List<Tweet> filtered2 = tweets.parallel().filter(tweet ->
>>> greetings.parallel().fold(() -> false, (Boolean b, String keyword) -> b ||
>>> tweet.text.contains(keyword), null)).into(new ArrayList<Tweet>());//
>>> greetings.parallel() fails with NPE when used with fold()
>>> 
>>> Exception in thread "main" java.lang.NullPointerException
>>>    at
>>> java.util.streams.ops.FoldOp$ReduceTask.onCompletion(FoldOp.java:123)
>>>    at
>>> 
>>> java.util.concurrent.CountedCompleter.tryComplete(CountedCompleter.java:389)
>>>    at java.util.streams.ops.AbstractTask.compute(AbstractTask.java:101)
>>>    at
>>> java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:435)
>>>    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:263)
>>>    at
>>> 
>>> java.util.concurrent.ForkJoinPool$WorkQueue.runSubtask(ForkJoinPool.java:994)
>>>    at
>>> java.util.concurrent.ForkJoinPool.tryHelpStealer(ForkJoinPool.java:1704)
>>>    at java.util.concurrent.ForkJoinPool.awaitJoin(ForkJoinPool.java:1833)
>>>    at java.util.concurrent.ForkJoinTask.doJoin(ForkJoinTask.java:345)
>>>    at java.util.concurrent.ForkJoinTask.join(ForkJoinTask.java:657)
>>>    at java.util.concurrent.ForkJoinPool.invoke(ForkJoinPool.java:2196)
>>>    at
>>> 
>>> java.util.streams.AbstractPipeline$ParallelImplPipelineHelper.invoke(AbstractPipeline.java:263)
>>>    at java.util.streams.ops.FoldOp.evaluateParallel(FoldOp.java:86)
>>>    at
>>> 
>>> java.util.streams.AbstractPipeline.evaluateParallel(AbstractPipeline.java:125)
>>>    at
>>> 
>>> java.util.streams.AbstractPipeline.evaluateParallel(AbstractPipeline.java:113)
>>>    at
>>> java.util.streams.AbstractPipeline.evaluate(AbstractPipeline.java:87)
>>>    at
>>> java.util.streams.AbstractPipeline.pipeline(AbstractPipeline.java:298)
>>>    at java.util.streams.ValuePipeline.fold(ValuePipeline.java:189)
>>>    at FilterTest$2.test(FilterTest.java:25)
>>>    at FilterTest$2.test(FilterTest.java:1)
>>>    at java.util.streams.ops.FilterOp$1.accept(FilterOp.java:62)
>>>    at java.util.streams.Streams$ArraySpliterator.into(Streams.java:711)
>>>    at
>>> 
>>> java.util.streams.ParallelPipelineHelper.into(ParallelPipelineHelper.java:57)
>>>    at
>>> 
>>> java.util.streams.AbstractPipeline$ParallelImplPipelineHelper.into(AbstractPipeline.java)
>>>    at
>>> java.util.streams.ops.TreeUtils$CollectorTask.doLeaf(TreeUtils.java:120)
>>>    at
>>> java.util.streams.ops.TreeUtils$CollectorTask.doLeaf(TreeUtils.java:98)
>>>    at java.util.streams.ops.AbstractTask.compute(AbstractTask.java:100)
>>>    at
>>> java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:435)
>>>    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:263)
>>>    at
>>> java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:975)
>>>    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1479)
>>>    at
>>> 
>>> java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:104)
>>> 
>>> -Arul
>>> 
>> 
>