haskell - recursivos - ¿Hay alguna forma de alinear una función recursiva?

recursividad

Question

Este es un seguimiento de mi pregunta anterior en la que pregunté por qué Stream Fusion no estaba funcionando en un determinado programa. Resulta que el problema era que ciertas funciones no estaban en línea, y una INLINE mejoró el rendimiento en aproximadamente 17x (¡lo que demuestra la importancia de la línea!).

Ahora, note que, en la pregunta original, incAll 64 llamadas de incAll a la vez. Ahora, supongamos que, en cambio, creo una función nTimes , que llama a una función repetidamente:

module Main where

import qualified Data.Vector.Unboxed as V

{-# INLINE incAll #-}
incAll :: V.Vector Int -> V.Vector Int
incAll = V.map (+ 1)

{-# INLINE nTimes #-}
nTimes :: Int -> (a -> a) -> a -> a
nTimes 0 f x = x
nTimes n f x = f (nTimes (n-1) f x)

main :: IO ()
main = do
  let size = 100000000 :: Int
  let array = V.replicate size 0 :: V.Vector Int
  print $ V.sum (nTimes 64 incAll array)

En este caso, solo agregar un pragma INLINE a nTimes no ayudará, porque AFAIK GHC no hace funciones recursivas en línea. ¿Hay algún truco para forzar a GHC a expandir nTimes en tiempo de compilación y, por lo tanto, recuperar el rendimiento esperado?

Answer 1

Hay un truco poco conocido que Andrés me ha dicho antes, en el que realmente puede obtener GHC para funciones recursivas en línea utilizando clases de tipo.

La idea es que en lugar de escribir una función normalmente donde se realiza una recursión estructural en un valor. Usted define su función utilizando clases de tipo y realiza una recursión estructural en un argumento de tipo. En este ejemplo, los números naturales a nivel de tipo.

GHC estará felizmente en línea con cada llamada recursiva y producirá un código eficiente ya que cada llamada recursiva es de un tipo diferente.

No comparé esto ni miré el núcleo, pero es notablemente más rápido.

{-# LANGUAGE DataKinds #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE PolyKinds #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Main where

import qualified Data.Vector.Unboxed as V

data Proxy a = Proxy

{-# INLINE incAll #-}
incAll :: V.Vector Int -> V.Vector Int
incAll = V.map (+ 1)

oldNTimes :: Int -> (a -> a) -> a -> a
oldNTimes 0 f x = x
oldNTimes n f x = f (oldNTimes (n-1) f x)

-- New definition

data N = Z | S N

class Unroll (n :: N) where
    nTimes :: Proxy n -> (a -> a) -> a -> a

instance Unroll Z where
    nTimes _ f x = x

instance Unroll n => Unroll (S n) where
    nTimes p f x =
        let Proxy :: Proxy (S n) = p
        in f (nTimes (Proxy :: Proxy n) f x)

main :: IO ()
main = do
  let size = 100000000 :: Int
  let array = V.replicate size 0 :: V.Vector Int
  print $ V.sum (nTimes (Proxy :: Proxy (S (S (S (S (S (S (S (S (S (S (S Z)))))))))))) incAll array)
  print $ V.sum (oldNTimes 11 incAll array)

Answer 2

No, pero puedes usar mejores funciones. No estoy hablando de V.map (+64) , lo que haría las cosas mucho más rápidas, pero sí de nTimes . Tenemos tres candidatos que ya hacen lo que hace nTimes :

{-# INLINE nTimesFoldr #-}
nTimesFoldr :: Int -> (a -> a) -> a -> a    
nTimesFoldr n f x = foldr (.) id (replicate n f) $ x

{-# INLINE nTimesIterate #-}
nTimesIterate :: Int -> (a -> a) -> a -> a    
nTimesIterate n f x = iterate f x !! n

{-# INLINE nTimesTail #-}
nTimesTail :: Int -> (a -> a) -> a -> a    
nTimesTail n f = go n
  where
    {-# INLINE go #-}
    go n x | n <= 0 = x
    go n x          = go (n - 1) (f x)

Todas las versiones tardan alrededor de 8 segundos, en comparación con los 40 segundos que tardan sus versiones. La versión de Joachim también lleva 8 segundos, por cierto. Tenga en cuenta que la versión iterate ocupa más memoria en mi sistema. Si bien hay un complemento de desenrollado para GHC, no se ha actualizado en los últimos cinco años (utiliza anotaciones personalizadas).

¿No desenrollar en absoluto?

Sin embargo, antes de que nos desesperemos, ¿qué tan bien intenta realmente GHC alinear todo? nTimesTail y nTimes 1 :

module Main where
import qualified Data.Vector.Unboxed as V

{-# INLINE incAll #-}
incAll :: V.Vector Int -> V.Vector Int
incAll = V.map (+ 1)

{-# INLINE nTimes #-}
nTimes :: Int -> (a -> a) -> a -> a    
nTimes n f = go n
  where
    {-# INLINE go #-}
    go n x | n <= 0 = x
    go n x          = go (n - 1) (f x)

main :: IO ()
main = do
  let size = 100000000 :: Int
  let array = V.replicate size 0 :: V.Vector Int
  print $ V.sum (nTimes 1 incAll array)

$ stack ghc --package vector -- -O2 -ddump-simpl -dsuppress-all SO.hs

main2 =
  case (runSTRep main3) `cast` ...
  of _ { Vector ww1_s9vw ww2_s9vx ww3_s9vy ->
  case $wgo 1 ww1_s9vw ww2_s9vx ww3_s9vy
  of _ { (# ww5_s9w3, ww6_s9w4, ww7_s9w5 #) ->

Podemos detenernos allí. $wgo es el go definido anteriormente. Incluso con 1 GHC no se desenrolla el bucle. Es inquietante ya que 1 es una constante.

Plantillas para el rescate.

Pero, por desgracia, no es todo para nada. Si los programadores de C ++ son capaces de hacer lo siguiente para las constantes de tiempo de compilación, ¿deberíamos nosotros?

template <int N>
struct Call{
    template <class F, class T>
    static T call(F f, T && t){
        return f(Call<N-1>::call(f,std::forward<T>(t)));
    }
};
template <>
struct Call<0>{
    template <class F, class T>
    static T call(F f, T && t){
        return t;
    }  
};

Y por supuesto, podemos con TemplateHaskell ^* :

-- Times.sh
{-# LANGUAGE TemplateHaskell #-}
module Times where

import Control.Monad (when)
import Language.Haskell.TH

nTimesTH :: Int -> Q Exp
nTimesTH n = do
  f <- newName "f"
  x <- newName "x"

  when (n <= 0) (reportWarning "nTimesTH: argument non-positive")

  let go k | k <= 0 = VarE x
      go k          = AppE (VarE f) (go (k - 1))
  return $ LamE [VarP f,VarP x] (go n)

¿Qué hace nTimesTH ? Crea una nueva función donde el primer nombre f se aplicará al segundo nombre x por un total de n veces. Ahora n necesita ser una constante de tiempo de compilación, lo que nos conviene, ya que el desenrollado de bucle solo es posible con constantes de tiempo de compilación:

$(nTimesTH 0) = /f x -> x
$(nTimesTH 1) = /f x -> f x
$(nTimesTH 2) = /f x -> f (f x)
$(nTimesTH 3) = /f x -> f (f (f x))
...

¿Funciona? Y es rapido ¿Qué tan rápido en comparación con nTimes ? Probemos otro main para eso:

-- SO.hs
{-# LANGUAGE TemplateHaskell #-}
module Main where
import Times
import qualified Data.Vector.Unboxed as V

{-# INLINE incAll #-}
incAll :: V.Vector Int -> V.Vector Int
incAll = V.map (+ 1)

{-# INLINE nTimes #-}
nTimes :: Int -> (a -> a) -> a -> a    
nTimes n f = go n
  where
    {-# INLINE go #-}
    go n x | n <= 0 = x
    go n x          = go (n - 1) (f x)

main :: IO ()
main = do
  let size = 100000000 :: Int
  let array = V.replicate size 0 :: V.Vector Int
  let vTH   = V.sum ($(nTimesTH 64) incAll array)
  let vNorm = V.sum (nTimes 64 incAll array)
  print $ vTH == vNorm

stack ghc --package vector -- -O2 SO.hs && SO.exe +RTS -t

True
<<ghc: 52000056768 bytes, 66 GCs, 400034700/800026736 avg/max bytes residency (2 samples), 1527M in use, 0.000 INIT (0.000 elapsed), 8.875 MUT (9.119 elapsed), 0.000 GC (0.094 elapsed) :ghc>>

Da el resultado correcto. ¿Qué tan rápido es? Usemos otro main otra vez:

main :: IO ()
main = do
  let size = 100000000 :: Int
  let array = V.replicate size 0 :: V.Vector Int
  print $ V.sum ($(nTimesTH 64) incAll array)

     800,048,112 bytes allocated in the heap                                         
           4,352 bytes copied during GC                                              
          42,664 bytes maximum residency (1 sample(s))                               
          18,776 bytes maximum slop                                                  
             764 MB total memory in use (0 MB lost due to fragmentation)             

                                     Tot time (elapsed)  Avg pause  Max pause        
  Gen  0         1 colls,     0 par    0.000s   0.000s     0.0000s    0.0000s        
  Gen  1         1 colls,     0 par    0.000s   0.049s     0.0488s    0.0488s        

  INIT    time    0.000s  (  0.000s elapsed)                                         
  MUT     time    0.172s  (  0.221s elapsed)                                         
  GC      time    0.000s  (  0.049s elapsed)                                         
  EXIT    time    0.000s  (  0.049s elapsed)                                         
  Total   time    0.188s  (  0.319s elapsed)                                         

  %GC     time       0.0%  (15.3% elapsed)                                           

  Alloc rate    4,654,825,378 bytes per MUT second                                   

  Productivity 100.0% of total user, 58.7% of total elapsed        

Bueno, compara eso con los 8s. Entonces, para un TL; DR : si tiene constantes de tiempo de compilación y desea crear y / o modificar su código en base a esas constantes, considere la Plantilla Haskell.

_{* Tenga en cuenta que este es mi primer código de Template Haskell que he escrito. Utilizar con cuidado. No uses una n demasiado grande, o podrías terminar con una función desordenada.}

Answer 3

No.

Usted podría escribir

{-# INLINE nTimes #-}
nTimes :: Int -> (a -> a) -> a -> a
nTimes n f x = go n
  where go 0 = x
        go n = f (go (n-1))

y GHC en línea nTimes , y probablemente se especialice en el recursivo, go a sus argumentos particulares en todos los incAll , pero no desenrollará el ciclo.