Grouping stream elements by their position - how to handle tail of stream ?

[Howard Lovatt]

Hi Brian,

This is an enquiry related to grouping or parsing an input stream, but
a little different.

In numerical programs common operations are matrix: transpose, other
reshaping, or 'sparsification'. I have seen these implemented on
parallel architectures as an indirection of indexes, something like:

matrix[indexTransformation[index]]

Has thought been given to such a capability for the stream library?

Thanks,

 -- Howard.

Sent from my iPad

On 03/02/2013, at 3:04 PM, Brian Goetz <brian.goetz at oracle.com> wrote:

> Look at the implementations in Collectors for a hint.  Obviously we owe
> some significant documentation, which unfortunately lags the code by too
> much.  But a Collector is basically a mutable foldl/inject; you provide
> a way to create a new mutable container, a way to add a single element
> to it, and a way to combine two mutable containers.  The framework
> decomposes the source, creates a bunch of little containers at the
> leaves, and them goes up the tree, merging.
>
> Dumping the elements into an ArrayList maps easily to Collector:
>
>  collect(ArrayList::new, ArrayList::add, ArrayList::addAll)
>
> This parallelizes, as we can collect each chunk into a list, and then
> merge the lists up the tree with addAll.
>
> The "how do I get the last element" problem goes away, since your data
> structure can explicitly maintain the state of whether the chunk ends
> with a balanced pair or an extra.
>
> But, you need to calculate BOTH p0 and p1 for each chunk (except maybe
> the left-spine chunks), because you don't know until the end which will
> have the right parity.
>
> On 2/2/2013 7:49 PM, Lattie wrote:
>>> Now, that said, here's a parallel algorithm you can implement with
>>> collect(), though you may find the performance overhead too offensive.
>>> Basically, for each chunk of the input t1..tn, you compute two possible
>>> answers:
>>>
>>> p0 = (t1,t2), (t3,t4), ..., maybe leftover tn
>>
>> This is what I want... the ability to compute p0.  I was not able to
>> figure out how to use collect() to do this though, and so I went down
>> the road of explode() with a stateful lambda... can someone point me
>> to any tutorial or info on how to properly use collect()?
>>
>> TIA
>>
>>
>> On Sat, Feb 2, 2013 at 1:55 PM, Brian Goetz <brian.goetz at oracle.com> wrote:
>>> The glass is half full, or half empty, depending on your disposition.
>>>
>>> While its trivial to write a version of explode() that remembers the
>>> last element seen, and either emits nothing or a pair (you have a little
>>> fixup at the end to deal with, but that's easy too, just annoying), once
>>> you start writing such stateful lambdas, you are tying yourself to
>>> sequential execution in a very error-prone way.  The type system can't
>>> record this assumption, so when someone helpfully later adds a
>>> .parallel() somewhere, your code will silently turn to sewage.  So,
>>> don't ever do this.
>>>
>>> Too see why the obvious algorithm is sequential only, consider a
>>> decomposition of a data source with a spliterator.  In most cases, we
>>> don't necessarily know the even-ness or odd-ness of the sum of sizes of
>>> all prior splits.  Which means we don't know whether the first element
>>> of any given split is the left element of some pair or the right element
>>> of some pair.
>>>
>>> You might say: I don't care about parallelism, I only care about
>>> sequential.
>>>
>>> To which we'd respond: fine, but we're not really interested in
>>> distorting the model to encourage that.  The Stream API is designed
>>> around operations that can be equally well performed in serial or
>>> parallel.  There are no "serial only" operations supported, and that was
>>> an explicit design choice.  Passing state-holding lambdas to these
>>> methods is likely to be a spec violation.  We can't enforce it, any more
>>> than we can enforce the lack of data races, but bear in mind that if you
>>> do this, you're writing broken code, and asking for trouble.
>>>
>>> Now, that said, here's a parallel algorithm you can implement with
>>> collect(), though you may find the performance overhead too offensive.
>>> Basically, for each chunk of the input t1..tn, you compute two possible
>>> answers:
>>>
>>> p0 = (t1,t2), t3,t4), ..., maybe leftover tn
>>> p1 = t1, (t2, t3), (t4, t5), ... maybe leftover tn
>>>
>>> then you combine these pairs chunks in the mostly obvious way (observe
>>> that p0 has trailing element = !(p1 has trailing element)).  Then when
>>> you get to the top of the tree, you take the one that doesn't have an
>>> orphaned initial element, and toss the other.  Basically you're doing 2x
>>> work for each input element, but it parallelizes fairly cleanly.
>>>
>>> Such an algorithm is ideally suited to collect(), because a Collector is
>>> a representation of a catamorphism, which the above transformation is.
>>> Because the combination logic is associative, it parallelizes cleanly
>>> without needing any nasty stateful lambdas.
>>>
>>> When we expose the Op APIs and you can write your own
>>> decomposition-aware operations, you'll have another option: write a
>>> StatefulOp.  This won't be an option for 8, but once we do, its easy
>>> enough, and it will let you avoid the extra overhead by creating a
>>> custom operation.
>>>
>>> So, to sum up:
>>>   - If you're looking for a one-liner, you'll be disappointed;
>>>   - If you convince yourself "I would never need parallelism here" and
>>> write the obvious unsafe stateful lambda, please write your manager a
>>> letter first explaining how your incompetence is putting the reliability
>>> of his product in jeopardy;
>>>   - If you're willing to write the above collector, you'll likely be
>>> happy enough with the result.
>>>
>>>
>>> On 2/2/2013 3:35 PM, Boaz Nahum wrote:
>>>> Hi.
>>>>
>>>> I looking in Stream interface for a way to convert a stream of
>>>>     {  t1, t2, .. Tn }
>>>> to
>>>>     { {t1,t2}, ... {Tn-1, Tn} }
>>>> or
>>>>     { {t1,t2}, ... {Tn, null}}
>>>>
>>>> Lets assume {t1, t2} are aggeraget by Pair<T,T>
>>>>
>>>>
>>>>
>>>> So I tried to use explode:
>>>>
>>>> *        Stream<Integer> si = Arrays.asList(1, 2, 3, 4, 5).stream();
>>>>
>>>>          Stream<Pair<Integer, Integer>> spi = si.sequential().explode(new
>>>> BiConsumer<Stream.Downstream<Pair<Integer, Integer>>, Integer>() {
>>>>
>>>>              Integer firstValue;
>>>>
>>>>              @Override
>>>>              public void accept(Stream.Downstream<Pair<Integer, Integer>>
>>>> pairDownstream, Integer v) {
>>>>
>>>>                  if (firstValue == null) {
>>>>                      firstValue = v;
>>>>                  } else {
>>>>
>>>>                      pairDownstream.send(new Pair<>(firstValue, v));
>>>>                      firstValue = null;
>>>>
>>>>                  }
>>>>              }
>>>>
>>>>          });
>>>> *
>>>>
>>>> But I didn't find a way to add the tail of input stream {.., 5} to the new
>>>> stream { ,,, {5, null}}.
>>>>
>>>> Of-course this only simplified example of more general problem I have.
>>>>
>>>> Thanks
>>>> Boaz
>