Cancelation -- use cases

[Sam Pullara]

I'm not a big fan of putting global conditions / side effects into the
stream operations. Obviously you can do the same thing with my gate
suggestion and can even add arbitrary cancellation:

AtomicBoolean until = new AtomicBoolean(true);

       Streams.iterate(from, i -> i + 1)  // sequential
                    .filter(i -> isPrime(i))
                    .until(() -> until.get())).
                    .forEach(i -> {
                        chm.put(i, true);
                    };

... until.set(false). But without the close() propogating up the chain, I'm
not sure how to recover resources from the Stream except through GC
finalizers. We can't simply exhaust the stream as it could be of infinite
length or very expensive to complete.

On Mon, Dec 31, 2012 at 8:56 AM, Remi Forax <forax at univ-mlv.fr> wrote:

> I've trouble to understood the difference with:
>
>        Streams.iterate(from, i -> i + 1)  // sequential
>                     .filter(i -> isPrime(i))
>                     .until(() -> System.currentTimeMillis() >= start+num)).
>                     .forEachUntil(i -> {
>                         chm.put(i, true);
>                     };
>
> Rémi
>
>
> On 12/31/2012 04:53 AM, Brian Goetz wrote:
>
>> Here's a lower-complexity version of cancel, that still satisfies (in
>> series or in parallel) use cases like the following:
>>
>> >   - Find the best possible move after thinking for 5 seconds
>> >   - Find the first solution that is better than X
>> >   - Gather solutions until we have 100 of them
>>
>> without bringing in the complexity or time/space overhead of dealing with
>> encounter order.
>>
>> Since the forEach() operation works exclusively on the basis of
>> temporal/arrival order rather than spatial/encounter order (elements are
>> passed to the lambda in whatever order they are available, in whatever
>> thread they are available), we could make a canceling variant of forEach:
>>
>>   .forEachUntil(Block<T> sink, BooleanSupplier until)
>>
>> Here, there is no confusion about what happens in the ordered case, no
>> need to buffer elements, etc.  Elements flow into the block until the
>> termination condition transpires, at which point there are no more splits
>> and existing splits dispense no more elements.
>>
>> I implemented this (it was trivial) and wrote a simple test program to
>> calculate primes sequentially and in parallel, counting how many could be
>> calculated in a fixed amount of time, starting from an infinite generator
>> and filtering out composites:
>>
>>             Streams.iterate(from, i -> i + 1)  // sequential
>>                     .filter(i -> isPrime(i))
>>                     .forEachUntil(i -> {
>>                         chm.put(i, true);
>>                     }, () -> System.currentTimeMillis() >= start+num);
>>
>> vs
>>
>>             Streams.iterate(from, i -> i+1)    // parallel
>>                     .parallel()
>>                     .filter(i -> isPrime(i))
>>                     .forEachUntil(i -> {
>>                         chm.put(i, true);
>>                     }, () -> System.currentTimeMillis() >= start+num);
>>
>> On a 4-core Q6600 system, in a fixed amount of time, the parallel version
>> gathered ~3x as many primes.
>>
>> In terms of being able to perform useful computations on infinite
>> streams, this seems a pretty attractive price-performer; lower spec and
>> implementation complexity, and covers many of the use cases which would
>> otherwise be impractical to attack with the stream approach.
>>
>>
>>
>> On 12/28/2012 11:20 AM, Brian Goetz wrote:
>>
>>> I've been working through some alternatives for cancellation support in
>>> infinite streams.  Looking to gather some use case background to help
>>> evaluate the alternatives.
>>>
>>> In the serial case, the "gate" approach works fine -- after some
>>> criteria transpires, stop sending elements downstream.  The pipeline
>>> flushes the elements it has, and completes early.
>>>
>>> In the parallel unordered case, the gate approach similarly works fine
>>> -- after the cancelation criteria occurs, no new splits are created, and
>>> existing splits dispense no more elements.  The computation similarly
>>> quiesces after elements currently being processed are completed,
>>> possibly along with any up-tree merging to combine results.
>>>
>>> It is the parallel ordered case that is tricky.  Supposing we partition
>>> a stream into
>>>    (a1,a2,a3), (a4,a5,a6)
>>>
>>> And suppose further we happen to be processing a5 when the bell goes
>>> off.  Do we want to wait for all a_i, i<5, to finish before letting the
>>> computation quiesce?
>>>
>>> My gut says: for the things we intend to cancel, most of them will be
>>> order-insensitive anyway.  Things like:
>>>
>>>   - Find the best possible move after thinking for 5 seconds
>>>   - Find the first solution that is better than X
>>>   - Gather solutions until we have 100 of them
>>>
>>> I believe the key use case for cancelation here will be when we are
>>> chewing on potentially infinite streams of events (probably backed by
>>> IO) where we want to chew until we're asked to shut down, and want to
>>> get as much parallelism as we can cheaply.  Which suggests to me the
>>> intersection between order-sensitive stream pipelines and cancelable
>>> stream pipelines is going to be pretty small indeed.
>>>
>>> Anyone want to add to this model of use cases for cancelation?
>>>
>>>
>