In che modo concatenare i flussi con pigrizia?

Question

Sto tentando di implementare uno stream che utilizza un'altra istanza di se stesso nella sua implementazione. Lo stream ha alcuni elementi costanti anteposti (con IntStream.concat) ad esso, quindi questo dovrebbe funzionare fintanto che il flusso concatenato crea la parte non costante pigramente. Penso che l'utilizzo di StreamSupport.intStream overload taking a Supplier con IntStream.concat (che è "creates a lazily concatenated stream") dovrebbe essere abbastanza pigro da creare solo il secondo spliterator quando gli vengono richiesti elementi, ma anche la creazione dello stream (non la sua valutazione) fa traboccare lo stack. Come posso pigramente concatenare i flussi?

---

che sto tentando di porto in streaming primo numero di setaccio da this answer in Java. Questo setaccio utilizza un'altra istanza di sé (ps = postponed_sieve() nel codice Python). Se mi rompo i primi quattro elementi costanti (yield 2; yield 3; yield 5; yield 7;) nella loro flusso, è facile da implementare il generatore come spliterator:

/** 
* based on https://stackoverflow.com/a/10733621/3614835 
*/ 
static class PrimeSpliterator extends Spliterators.AbstractIntSpliterator { 
    private static final int CHARACTERISTICS = Spliterator.DISTINCT | Spliterator.IMMUTABLE | Spliterator.NONNULL | Spliterator.ORDERED | Spliterator.SORTED; 
    private final Map<Integer, Supplier<IntStream>> sieve = new HashMap<>(); 
    private final PrimitiveIterator.OfInt postponedSieve = primes().iterator(); 
    private int p, q, c = 9; 
    private Supplier<IntStream> s; 
    PrimeSpliterator() { 
     super(105097564 /* according to Wolfram Alpha */ - 4 /* in prefix */, 
       CHARACTERISTICS); 
     //p = next(ps) and next(ps) (that's Pythonic?) 
     postponedSieve.nextInt(); 
     this.p = postponedSieve.nextInt(); 
     this.q = p*p; 
    } 

    @Override 
    public boolean tryAdvance(IntConsumer action) { 
     for (; c > 0 /* overflow */; c += 2) { 
      Supplier<IntStream> maybeS = sieve.remove(c); 
      if (maybeS != null) 
       s = maybeS; 
      else if (c < q) { 
       action.accept(c); 
       return true; //continue 
      } else { 
       s =() -> IntStream.iterate(q+2*p, x -> x + 2*p); 
       p = postponedSieve.nextInt(); 
       q = p*p; 
      } 
      int m = s.get().filter(x -> !sieve.containsKey(x)).findFirst().getAsInt(); 
      sieve.put(m, s); 
     } 
     return false; 
    } 
} 

Il mio primo tentativo di numeri primi() restituisce un IntStream concatenando un flusso costante con un nuovo PrimeSpliterator:

public static IntStream primes() { 
    return IntStream.concat(IntStream.of(2, 3, 5, 7), 
      StreamSupport.intStream(new PrimeSpliterator())); 
} 

numeri primi di chiamata() si traduce in una StackOverflowError perché i numeri primi() crea un'istanza sempre un PrimeSpliterator, ma campo di inizializzazione del PrimeSpliterator chiama sempre numeri primi(). Tuttavia, c'è un sovraccarico di StreamSupport.intStream che prende un fornitore, che dovrebbe consentire la creazione pigramente il PrimeSpliterator:

public static IntStream primes() { 
    return IntStream.concat(IntStream.of(2, 3, 5, 7), 
      StreamSupport.intStream(PrimeSpliterator::new, PrimeSpliterator.CHARACTERISTICS, false)); 
} 

Tuttavia, ho invece ottengo uno StackOverflowError con un backtrace diversa (tagliato, come si ripete). Si noti che la ricorsione è interamente nella chiamata a primes() - l'operazione terminale iterator() non viene mai invocata su un flusso restituito.

Exception in thread "main" java.lang.StackOverflowError 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator$OfInt.<init>(StreamSpliterators.java:582) 
    at java.util.stream.IntPipeline.lazySpliterator(IntPipeline.java:155) 
    at java.util.stream.IntPipeline$Head.lazySpliterator(IntPipeline.java:514) 
    at java.util.stream.AbstractPipeline.spliterator(AbstractPipeline.java:352) 
    at java.util.stream.IntPipeline.spliterator(IntPipeline.java:181) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 
    at com.jeffreybosboom.projecteuler.util.Primes$PrimeSpliterator.<init>(Primes.java:32) 
    at com.jeffreybosboom.projecteuler.util.Primes$$Lambda$1/834600351.get(Unknown Source) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.get(StreamSpliterators.java:513) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.estimateSize(StreamSpliterators.java:536) 
    at java.util.stream.Streams$ConcatSpliterator.<init>(Streams.java:713) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:789) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:785) 
    at java.util.stream.Streams$ConcatSpliterator$OfInt.<init>(Streams.java:819) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 
    at com.jeffreybosboom.projecteuler.util.Primes$PrimeSpliterator.<init>(Primes.java:32) 
    at com.jeffreybosboom.projecteuler.util.Primes$$Lambda$1/834600351.get(Unknown Source) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.get(StreamSpliterators.java:513) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.estimateSize(StreamSpliterators.java:536) 
    at java.util.stream.Streams$ConcatSpliterator.<init>(Streams.java:713) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:789) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:785) 
    at java.util.stream.Streams$ConcatSpliterator$OfInt.<init>(Streams.java:819) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 

Come posso concatenare i flussi abbastanza pigramente per permettere un flusso di usare un'altra copia di se stesso nella sua attuazione?

Answer 1

Il tuo apparentemente suppone che l'API Streams estende le sue garanzie di pigrizia anche alla istanziazione di spliterators; questo non è corretto Si aspetta di essere in grado di creare un'istanza dello spliterator del flusso in qualsiasi momento prima che inizi il consumo effettivo, ad esempio solo per scoprire le caratteristiche del flusso e le dimensioni segnalate. Il consumo inizia solo invocando trySplit, tryAdvance o forEachRemaining.

Tenendo presente questo, si sta inizializzando il setaccio posticipato prima del necessario. Non è possibile utilizzare nessuno dei risultati fino alla parte else if in tryAdvance. Quindi spostare il codice per l'all'ultimo momento che dà la correttezza:

@Override 
public boolean tryAdvance(IntConsumer action) { 
    for (; c > 0 /* overflow */; c += 2) { 
     Supplier<IntStream> maybeS = sieve.remove(c); 
     if (maybeS != null) 
      s = maybeS; 
     else { 
      if (postponedSieve == null) { 
       postponedSieve = primes().iterator(); 
       postponedSieve.nextInt(); 
       this.p = postponedSieve.nextInt(); 
       this.q = p*p; 
      } 
      if (c < q) { 
       action.accept(c); 
       return true; //continue 

penso che, con questo cambiamento, anche il vostro primo tentativo di primes() dovrebbe funzionare.

Se si vuole rimanere con il vostro approccio attuale, si potrebbe coinvolgere il seguente idioma:

Stream.<Supplier<IntStream>>of(
()->IntStream.of(2, 3, 5, 7), 
()->intStream(new PrimeSpliterator())) 
.flatMap(Supplier::get); 

È possibile che questo ti dà tanto la pigrizia di cui hai bisogno.