listas - recorrer lista en scala

Recorriendo listas y secuencias con una función que devuelve un futuro

Question

Introducción

El Future de Scala ( nuevo en 2.10 y ahora 2.9.3 ) es un funtor aplicativo, lo que significa que si tenemos un tipo F transitable , podemos tomar un F[A] y una función A => Future[B] y convertirlos en un Future[F[B]] .

Esta operación está disponible en la biblioteca estándar como Future.traverse . Scalaz 7 también proporciona un traverse más general que podemos usar aquí si importamos la instancia del funtor aplicativo para el Future desde la biblioteca scalaz-contrib .

Estos dos métodos traverse comportan de manera diferente en el caso de las corrientes. El recorrido estándar de la biblioteca consume el flujo antes de regresar, mientras que Scalaz devuelve el futuro inmediatamente :

import scala.concurrent._
import ExecutionContext.Implicits.global

// Hangs.
val standardRes = Future.traverse(Stream.from(1))(future(_))

// Returns immediately.
val scalazRes = Stream.from(1).traverse(future(_))

También hay otra diferencia, como observa Leif Warner here . La traverse la biblioteca estándar inicia todas las operaciones asíncronas inmediatamente, mientras que Scalaz inicia la primera, espera a que se complete, inicia la segunda, la espera, y así sucesivamente.

Diferentes comportamientos para arroyos.

Es bastante fácil mostrar esta segunda diferencia al escribir una función que se suspenderá durante unos segundos para el primer valor de la secuencia:

def howLong(i: Int) = if (i == 1) 10000 else 0

import scalaz._, Scalaz._
import scalaz.contrib.std._

def toFuture(i: Int)(implicit ec: ExecutionContext) = future {
  printf("Starting %d!/n", i)
  Thread.sleep(howLong(i))
  printf("Done %d!/n", i)
  i
}

Ahora Future.traverse(Stream(1, 2))(toFuture) imprimirá lo siguiente:

Starting 1!
Starting 2!
Done 2!
Done 1!

Y la versión Scalaz ( Stream(1, 2).traverse(toFuture) ):

Starting 1!
Done 1!
Starting 2!
Done 2!

Lo que probablemente no es lo que queremos aquí.

¿Y para listas?

Por extraño que parezca, los dos recorridos se comportan igual a este respecto en las listas; Scalaz''s no espera a que se complete un futuro antes de comenzar el siguiente.

Otro futuro

Scalaz también incluye su propio paquete concurrent con su propia implementación de futuros. Podemos usar el mismo tipo de configuración anterior:

import scalaz.concurrent.{ Future => FutureZ, _ }

def toFutureZ(i: Int) = FutureZ {
  printf("Starting %d!/n", i)
  Thread.sleep(howLong(i))
  printf("Done %d!/n", i)
  i
}

Y luego obtenemos el comportamiento de Scalaz en las secuencias para las listas , así como las secuencias:

Starting 1!
Done 1!
Starting 2!
Done 2!

Quizás menos sorprendente, al atravesar un flujo infinito todavía se devuelve de inmediato.

Pregunta

En este punto, realmente necesitamos una tabla para resumir, pero una lista tendrá que hacer:

• Transmisiones con el recorrido de la biblioteca estándar: consumir antes de regresar; No esperes a cada futuro.
• Streams con Scalaz transversal: volver inmediatamente; espere a que se complete cada futuro.
• Scalaz futuros con arroyos: volver de inmediato; espere a que se complete cada futuro.

Y:

• Listas con recorrido de la biblioteca estándar: no espere.
• Listas con el recorrido de Scalaz: no esperes.
• Futuros de Scalaz con listas: espere a que se complete cada futuro.

¿Tiene esto algún sentido? ¿Existe un comportamiento "correcto" para esta operación en listas y secuencias? ¿Hay alguna razón por la que el comportamiento "más asíncrono", es decir, no consuma la colección antes de regresar, y no espere a que se complete cada futuro antes de pasar al siguiente, no se representa aquí?

Answer 1

No puedo contestar todo, pero trato de algunas partes:

¿Hay alguna razón por la que el comportamiento "más asíncrono", es decir, no consuma la colección antes de regresar, y no espere a que se complete cada futuro antes de pasar al siguiente, no se representa aquí?

Si tiene cálculos dependientes y un número limitado de subprocesos, puede experimentar interbloqueos. Por ejemplo, tiene dos futuros dependiendo de un tercero (los tres en la lista de futuros) y solo dos subprocesos, puede experimentar una situación en la que los dos primeros futuros bloquean los dos subprocesos y el tercero nunca se ejecuta. (Por supuesto, si el tamaño de su grupo es uno, es decir, si ejecuta un cálculo después del otro, puede obtener situaciones similares)

Para resolver esto, necesita un hilo por futuro, sin ninguna limitación. Esto funciona para pequeñas listas de futuros, pero no para grandes. Entonces, si ejecuta todo en paralelo, obtendrá una situación en la que se ejecutarán pequeños ejemplos en todos los casos y uno más grande se interrumpirá. (Ejemplo: las pruebas de desarrollador funcionan bien, puntos muertos de producción).

¿Existe un comportamiento "correcto" para esta operación en listas y secuencias?

Creo que es imposible con los futuros. Si sabe algo más de las dependencias, o cuando sabe con seguridad que los cálculos no se bloquearán, podría ser posible una solución más concurrente. Pero la ejecución de listas de futuros me busca "roto por diseño". La mejor solución parece ser una, que ya fallará para pequeños ejemplos de interbloqueos (es decir, ejecutar un Futuro tras otro).

Futuros de Scalaz con listas: espere a que se complete cada futuro.

Creo que Scalaz usa internamente las comprensiones para el recorrido. Con las comprensiones, no se garantiza que los cálculos sean independientes. Así que supongo que Scalaz está haciendo lo correcto aquí con las comprensiones: haciendo un cálculo después del otro. En el caso de futuros, esto siempre funcionará, dado que tiene subprocesos ilimitados en su sistema operativo.

En otras palabras: usted ve solo un artefacto de cómo funcionan las comprensiones (debe).

Espero que esto tenga algún sentido.

Answer 2

Si entiendo la pregunta correctamente, creo que realmente se reduce a la semántica de las listas de streams vs.

atravesar una lista hace lo que esperaríamos de los documentos:

Transforma una TraversableOnce[A] en un Future[TraversableOnce[B]] usando la función provista A => Future[B] . Esto es útil para realizar un mapa paralelo. Por ejemplo, para aplicar una función a todos los elementos de una lista en paralelo:

con los flujos, depende del desarrollador decidir cómo quieren que funcione, ya que depende de un mayor conocimiento del flujo que tiene el compilador (los flujos pueden ser infinitos, pero el sistema de tipos no lo sabe). Si mi flujo está leyendo líneas de un archivo, quiero consumirlo primero, ya que encadenar los futuros línea por línea no paralizaría las cosas. En este caso querría el enfoque paralelo.

por otro lado, si mi flujo es una lista infinita que genera enteros secuenciales y busca el primer número primo mayor que un número grande, sería imposible consumir el flujo primero en un barrido (se requeriría el enfoque Encadenado del Future probablemente querría ejecutar lotes de la secuencia).

En lugar de tratar de encontrar una manera canónica de manejar esto, me pregunto si faltan tipos que ayuden a hacer más explícitos los diferentes casos.