haskell - Marcos para representar el procesamiento de datos como una tubería

functional-programming pipe

Question

La mayor parte del procesamiento de datos se puede visualizar como una tubería de componentes, la salida de uno se alimenta de la entrada de otro. Una tubería de procesamiento típica es:

reader | handler | writer

Como una hoja para comenzar esta discusión, consideremos una implementación orientada a objetos de esta tubería donde cada segmento es un objeto. El objeto de handler contiene referencias a los objetos de reader y writer y tiene un método de run que tiene el siguiente aspecto:

define handler.run:
  while (reader.has_next) {
    data = reader.next
    output = ...some function of data...
    writer.put(output)
  }

Esquemáticamente las dependencias son:

reader <- handler -> writer

Ahora supongamos que quiero interponer un nuevo segmento de canalización entre el lector y el manejador:

reader | tweaker | handler | writer

Nuevamente, en esta implementación de OO, tweaker sería un envoltorio alrededor del objeto reader , y los métodos de tweaker podrían parecer algo así (en algún código pseudo-imperativo):

define tweaker.has_next:
  return reader.has_next

define tweaker.next:
  value = reader.next
  result = ...some function of value...
  return result

Estoy encontrando que esto no es una abstracción muy composable. Algunos temas son:

1. tweaker solo se puede usar en el lado izquierdo del handler , es decir, no puedo usar la implementación anterior de tweaker para formar este tweaker :

   lector manejador tweaker escritor

2. Me gustaría explotar la propiedad asociativa de tuberías, para que esta tubería:

   lector manejador escritor

podría expresarse como:

reader | p

donde p es el handler | writer tubería handler | writer handler | writer En esta implementación OO tendría que crear una instancia parcial del objeto handler

1. Algo de una reexpresión de (1), los objetos deben saber si "empujan" o "tiran" datos.

Estoy buscando un marco (no necesariamente OO) para crear tuberías de procesamiento de datos que aborden estos problemas.

He etiquetado esto con Haskell y la functional programming porque creo que los conceptos de programación funcional podrían ser útiles aquí.

Como objetivo, sería bueno poder crear una tubería como esta:

                     handler1
                   /          /
reader | partition              writer
                   /          /
                     handler2

Para cierta perspectiva, las tuberías de shell de Unix resuelven muchos de estos problemas con las siguientes decisiones de implementación:

1. Los componentes de Pipeline se ejecutan de forma asíncrona en procesos separados

2. Los objetos de tubería median los datos que pasan entre "empujadores" y "tiradores"; es decir, bloquean a los escritores que escriben datos demasiado rápido y los lectores que intentan leer demasiado rápido.

3. Utiliza conectores especiales < y > para conectar componentes pasivos (es decir, archivos) a la tubería

Estoy especialmente interesado en los enfoques que no utilizan el subprocesamiento o el paso de mensajes entre los agentes. Tal vez sea la mejor manera de hacer esto, pero me gustaría evitar los hilos si es posible.

¡Gracias!

Answer 1

El paquete del enumerador para Haskell es un buen marco para esto. Define tres tipos de objetos:

1. Enumeradores que producen datos en trozos.
2. Iteratees que consumen trozos de datos y devuelven un valor después de consumir lo suficiente.
3. Enumerados que se sientan en el medio de la tubería. Consumen trozos y producen trozos, posiblemente con efectos secundarios.

Estos tres tipos de objetos se componen en una tubería de procesamiento de flujo, e incluso puede tener múltiples enumeradores e iteraciones en una tubería (cuando uno termina, el siguiente toma su lugar). Puede ser complicado escribir uno de estos objetos desde cero, pero hay muchos combinadores que se pueden usar para convertir funciones regulares en procesadores de flujo de datos. Por ejemplo, esta tubería lee todos los caracteres de stdin, los convierte a mayúsculas con la función toUpper , luego los escribe en stdout:

ET.enumHandle stdin $$ ET.map toUpper =$ ET.iterHandle stdout

donde el módulo Data.Enumerator.Text ha sido importado como ET .

Answer 2

El marco de Yesod hace uso de una biblioteca de tuberías de Haskell en la forma del paquete de conduit .

Answer 3

Gracias a la perezosa evaluación , podemos expresar las tuberías en términos de composiciones de funciones ordinarias en Haskell. Aquí un ejemplo que calcula la longitud máxima de una línea en un archivo:

main = interact (show . maximum . map length . lines)

Todo aquí es una función ordinaria, como por ejemplo

lines :: String -> [String]

pero gracias a la evaluación perezosa, estas funciones solo procesan la entrada de forma incremental y solo la cantidad necesaria, tal como lo haría un conducto UNIX.

Answer 4

Sí, las arrows son casi seguramente tu hombre.

Sospecho que eres bastante nuevo en Haskell, solo por el tipo de cosas que dices en tu pregunta. Las flechas probablemente parecerán bastante abstractas, especialmente si lo que buscas es un "marco". Sé que me tomó un tiempo realmente asimilar lo que estaba pasando con las flechas.

Así que puedes mirar esa página y decir "sí, que se parece a lo que quiero", y luego te encuentras perdido en cuanto a cómo comenzar a usar flechas para resolver el problema. Así que aquí hay un poco de orientación para que sepas lo que estás viendo.

Las flechas no resolverán tu problema. En su lugar, te dan un idioma que puedes usar en el que expresas tu problema. Puede encontrar que alguna flecha predefinida hará el trabajo, quizás una flecha kleisli, pero al final del día usted querrá implementar una flecha (las predefinidas simplemente le brindan formas fáciles de implementarlas) que expresan lo que quiere decir con un "procesador de datos". Como ejemplo casi trivial, supongamos que desea implementar sus procesadores de datos con funciones simples. Usted escribiría:

newtype Proc a b = Proc { unProc :: a -> b }

-- I believe Arrow has recently become a subclass of Category, so assuming that.

instance Category Proc where
    id = Proc (/x -> x)
    Proc f . Proc g = Proc (/x -> f (g x))

instance Arrow Proc where
    arr f = Proc f
    first (Proc f) = Proc (/(x,y) -> (f x, y))

Esto le da la maquinaria para usar los diversos combinadores de flecha (***) , (&&&) , (>>>) , etc., así como la notación de flecha que es bastante agradable si está haciendo cosas complejas. Entonces, como señala Daniel Fischer en el comentario, la tubería que describió en su pregunta se podría componer de la siguiente manera:

reader >>> partition >>> (handler1 *** handler2) >>> writer

Pero lo bueno es que depende de usted lo que quiere decir con un procesador. Es posible implementar lo que mencionó acerca de cada procesador forking un hilo de una manera similar, utilizando un tipo de procesador diferente:

newtype Proc'' a b = Proc (Source a -> Sink b -> IO ())

Y luego implementar adecuadamente los combinadores.

Así que eso es lo que está viendo: un vocabulario para hablar sobre los procesos de composición, que tiene un poco de código para reutilizar, pero principalmente ayudará a guiar su pensamiento a medida que implementa estos combinadores para la definición de procesador que sea útil en su dominio. .

Uno de mis primeros proyectos no triviales de Haskell fue implementar una flecha para el entrelazamiento cuántico ; ese proyecto fue el que me hizo comenzar a entender realmente la forma de pensar de Haskell, un punto de inflexión importante en mi carrera de programación. Tal vez este proyecto tuyo haga lo mismo por ti? :-)