Elaborazione funzionale di flussi Scala senza errori OutOfMemory

Question

È possibile applicare la programmazione funzionale a flussi Scala in modo che lo stream venga elaborato in modo sequenziale, ma la parte già elaborata del flusso può essere eliminata?

Per esempio, ho definire un Stream che contiene i numeri da start a end:

def fromToStream(start: Int, end: Int) : Stream[Int] = { 
    if (end < start) Stream.empty 
    else start #:: fromToStream(start+1, end) 
} 

Se io riassumere i valori in uno stile funzionale:

println(fromToStream(1,10000000).reduceLeft(_+_)) 

ottengo una OutOfMemoryError - forse dal momento che lo stackframe della chiamata a reduceLeft contiene un riferimento al capo dello stream. Ma se faccio questo in stile iterativo, funziona:

var sum = 0 
for (i <- fromToStream(1,10000000)) { 
    sum += i 
} 

C'è un modo per fare questo in uno stile funzionale, senza ottenere un OutOfMemory?

UPDATE: Questo è stato a bug in scala che è stato risolto ora. Quindi questo è più o meno obsoleto ora.

Answer 1

Sì, è possibile. Il trucco consiste nell'usare metodi ricorsivi tail, in modo che il frame stack locale contenga l'unico riferimento all'istanza Stream. Poiché il metodo è ricorsivo in coda, il riferimento locale alla precedente Stream viene cancellato una volta che si richiama in modo ricorsivo, consentendo in tal modo al GC di raccogliere l'inizio dello Stream come si va.

Welcome to Scala version 2.9.0.r23459-b20101108091606 (Java HotSpot(TM) Server VM, Java 1.6.0_20). 
Type in expressions to have them evaluated. 
Type :help for more information. 

scala> import collection.immutable.Stream 
import collection.immutable.Stream 

scala> import annotation.tailrec 
import annotation.tailrec 

scala> @tailrec def last(s: Stream[Int]): Int = if (s.tail.isEmpty) s.head else last(s.tail) 
last: (s: scala.collection.immutable.Stream[Int])Int 

scala> last(Stream.range(0, 100000000))                    
res2: Int = 99999999 

Inoltre, è necessario assicurarsi che la cosa si passa al metodo last ha un solo riferimento in pila sopra. Se si memorizza un valore Stream in una variabile o un valore locale, questo non verrà raccolto automaticamente quando si chiama il metodo last, poiché il suo argomento non è l'unico riferimento rimasto a Stream. Il codice sottostante esaurisce la memoria.

scala> val s = Stream.range(0, 100000000)                   
s: scala.collection.immutable.Stream[Int] = Stream(0, ?)                

scala> last(s)                          
Exception in thread "main" java.lang.OutOfMemoryError: Java heap space            
     at sun.net.www.ParseUtil.encodePath(ParseUtil.java:84)              
     at sun.misc.URLClassPath$JarLoader.checkResource(URLClassPath.java:674)          
     at sun.misc.URLClassPath$JarLoader.getResource(URLClassPath.java:759)          
     at sun.misc.URLClassPath.getResource(URLClassPath.java:169)             
     at java.net.URLClassLoader$1.run(URLClassLoader.java:194)             
     at java.security.AccessController.doPrivileged(Native Method)            
     at java.net.URLClassLoader.findClass(URLClassLoader.java:190)            
     at java.lang.ClassLoader.loadClass(ClassLoader.java:307)              
     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:301)            
     at java.lang.ClassLoader.loadClass(ClassLoader.java:248)              
     at scala.tools.nsc.Interpreter$Request$$anonfun$onErr$1$1.apply(Interpreter.scala:978)      
     at scala.tools.nsc.Interpreter$Request$$anonfun$onErr$1$1.apply(Interpreter.scala:976)      
     at scala.util.control.Exception$Catch.apply(Exception.scala:80) 
     at scala.tools.nsc.Interpreter$Request.loadAndRun(Interpreter.scala:984)          
     at scala.tools.nsc.Interpreter.loadAndRunReq$1(Interpreter.scala:579)          
     at scala.tools.nsc.Interpreter.interpret(Interpreter.scala:599)            
     at scala.tools.nsc.Interpreter.interpret(Interpreter.scala:576) 
     at scala.tools.nsc.InterpreterLoop.reallyInterpret$1(InterpreterLoop.scala:472)        
     at scala.tools.nsc.InterpreterLoop.interpretStartingWith(InterpreterLoop.scala:515)       
     at scala.tools.nsc.InterpreterLoop.command(InterpreterLoop.scala:362) 
     at scala.tools.nsc.InterpreterLoop.processLine$1(InterpreterLoop.scala:243) 
     at scala.tools.nsc.InterpreterLoop.repl(InterpreterLoop.scala:249) 
     at scala.tools.nsc.InterpreterLoop.main(InterpreterLoop.scala:559) 
     at scala.tools.nsc.MainGenericRunner$.process(MainGenericRunner.scala:75) 
     at scala.tools.nsc.MainGenericRunner$.main(MainGenericRunner.scala:31) 
     at scala.tools.nsc.MainGenericRunner.main(MainGenericRunner.scala) 

In sintesi:

1. Utilizzare metodi di coda ricorsiva
2. li annotare come la coda-ricorsiva
3. quando li chiami, garantire che il loro argomento è l'unico riferimento alla Stream

MODIFICA:

Nota che questo funziona anche e non si traduce in un errore di memoria:

scala> def s = Stream.range(0, 100000000)             
s: scala.collection.immutable.Stream[Int] 

scala> last(s)                    
res1: Int = 99999999 

EDIT2:

E nel caso di reduceLeft che si richiedono, si dovrà definire un metodo di supporto con un argomento accumulatore per il risultato.

Per reduceLeft, è necessario un argomento di accumulatore, che è possibile impostare su un determinato valore utilizzando argomenti predefiniti. Un esempio semplificato:

scala> @tailrec def rcl(s: Stream[Int], acc: Int = 0): Int = if (s.isEmpty) acc else rcl(s.tail, acc + s.head) 
rcl: (s: scala.collection.immutable.Stream[Int],acc: Int)Int 

scala> rcl(Stream.range(0, 10000000)) 
res6: Int = -2014260032 

Answer 2

Si consiglia di guardare Scalaz ephemeral streams.

Answer 3

Quando ho iniziato a conoscere Stream ho pensato che fosse bello. Poi ho capito che Iterator è quello che voglio usare quasi tutto il tempo.

Nel caso in cui si ha bisogno Stream ma vuole fare reduceLeft lavoro:

fromToStream(1,10000000).toIterator.reduceLeft(_ + _) 

Se si tenta la linea di cui sopra, sarà spazzatura raccogliere bene. Ho scoperto che usare Stream è complicato in quanto è facile aggrapparsi alla testa senza rendersene conto. A volte il lib standard si aggrappa ad esso per te - in modi molto sottili.

Answer 4

Come risulta, questo è a bug nell'implementazione corrente di reduceLeft. Il problema è che riduciLeft chiama foldLeft, e quindi lo stackframe di reduceLeft mantiene un riferimento alla testa del flusso durante l'intera chiamata. foldLeft usa la ricorsione in coda per evitare questo problema. Confronta:

(1 to 10000000).toStream.foldLeft(0)(_+_) 
(1 to 10000000).toStream.reduceLeft(_+_) 

Questi sono semanticamente equivalenti. Nella versione 2.8.0 di Scala, la chiamata a foldLeft funziona, ma la chiamata a reduce Left genera un oggetto OutOfMemory. Se riduciLeft farebbe il suo lavoro, questo problema non si verificherebbe.