algorithm - Usando vectores para mejorar el rendimiento en Haskell

Question

Soy muy nuevo en Haskell, y tengo una pregunta sobre qué mejoras de rendimiento se pueden lograr al usar estructuras de datos impuras (mutables). Estoy tratando de juntar algunas cosas diferentes que he escuchado, así que tenga paciencia si mi terminología no es del todo correcta o si hay algunos errores pequeños.

Para concretar esto, considere el algoritmo de ordenación rápida (tomado de la wiki de Haskell).

quicksort :: Ord a => [a] -> [a]
quicksort [] = []
quicksort (p:xs) = (quicksort lesser) ++ [p] ++ (quicksort greater)
where
lesser = filter (< p) xs
greater = filter (>= p) xs

Esto no es "verdadero quicksort" Un algoritmo "verdadero" de ordenación rápida está en el lugar, y esto no lo es. Esto es muy ineficiente de la memoria.

Por otro lado, es posible usar vectores en Haskell para implementar una orden rápida in situ. Un ejemplo se da en esta respuesta de stackoverflow.

¿Cuánto más rápido es el segundo algoritmo que el primero? La notación Big O no ayuda aquí, porque la mejora del rendimiento será a partir del uso más eficiente de la memoria, no de tener un algoritmo mejor (¿verdad?). Me cansé de construir algunos casos de prueba por mi cuenta, pero me costó poner las cosas en marcha.

Una respuesta ideal daría una idea de lo que hace que el algoritmo de Haskell en el lugar sea más rápido en teoría, y una comparación ejemplar de los tiempos de ejecución en algunos conjuntos de datos de prueba.

Answer 1

Por otro lado, es posible usar vectores en Haskell para implementar una orden rápida in situ.

¿Cuánto más rápido es el segundo algoritmo que el primero?

Eso depende de la implementación, por supuesto. Como se puede ver a continuación, para listas no demasiado cortas, una clasificación en el lugar decente en un vector o matriz mutable es mucho más rápida que las listas de clasificación, incluso si se incluye el tiempo para la transformación de y a las listas ( y la conversión se completa) la mayor parte del tiempo ).

Sin embargo, los algoritmos de la lista producen resultados incrementales, mientras que los algoritmos de matriz / vector no producen ningún resultado antes de que se hayan completado, por lo tanto, las listas de clasificación aún pueden ser preferibles.

No sé exactamente qué hicieron mal los algoritmos de matriz / vector vinculados vinculados. Pero hicieron algo muy mal.

Para el código vectorial mutable, parece que usó vectores en caja , y fue polimórfico, ambos pueden tener un impacto significativo en el rendimiento, aunque el polimorfismo no debería importar si las funciones son {-# INLINABLE #-} .

Para el código IOUArray , bueno, parece divertido, pero lento. Utiliza un IORef , readArray y writeArray y no tiene ningún rigor obvio. Los tiempos abismales que toma no son demasiado sorprendentes, entonces.

Usando una traducción más directa del código C (monomorfo) usando un STUArray , con una envoltura para hacer que funcione en listas¹,

{-# LANGUAGE BangPatterns #-}
module STUQuickSort (stuquick) where
import Data.Array.Base (unsafeRead, unsafeWrite)
import Data.Array.ST
import Control.Monad.ST
stuquick :: [Int] -> [Int]
stuquick [] = []
stuquick xs = runST (do
let !len = length xs
arr <- newListArray (0,len-1) xs
myqsort arr 0 (len-1)
-- Can''t use getElems for large arrays, that overflows the stack, wth?
let pick acc i
| i < 0 = return acc
| otherwise = do
!v <- unsafeRead arr i
pick (v:acc) (i-1)
pick [] (len-1))
myqsort :: STUArray s Int Int -> Int -> Int -> ST s ()
myqsort a lo hi
| lo < hi = do
let lscan p h i
| i < h = do
v <- unsafeRead a i
if p < v then return i else lscan p h (i+1)
| otherwise = return i
rscan p l i
| l < i = do
v <- unsafeRead a i
if v < p then return i else rscan p l (i-1)
| otherwise = return i
swap i j = do
v <- unsafeRead a i
unsafeRead a j >>= unsafeWrite a i
unsafeWrite a j v
sloop p l h
| l < h = do
l1 <- lscan p h l
h1 <- rscan p l1 h
if (l1 < h1) then (swap l1 h1 >> sloop p l1 h1) else return l1
| otherwise = return l
piv <- unsafeRead a hi
i <- sloop piv lo hi
swap i hi
myqsort a lo (i-1)
myqsort a (i+1) hi
| otherwise = return ()

y una envoltura alrededor de un buen tipo (Introsort, no quicksort) en vectores sin caja,

module VSort where
import Data.Vector.Algorithms.Intro
import qualified Data.Vector.Unboxed as U
import Control.Monad.ST
vsort :: [Int] -> [Int]
vsort xs = runST (do
v <- U.unsafeThaw $ U.fromList xs
sort v
s <- U.unsafeFreeze v
return $ U.toList s)

Me pongo a veces más en línea con las expectativas ( Nota: para estos tiempos, la lista aleatoria se ha deepseq antes de llamar al algoritmo de clasificación. Sin eso, la conversión a un STUArray sería mucho más lenta, ya que primero evaluaría una lista larga de thunks para determinar la longitud. La conversión fromList del paquete vector no sufre este problema. Mover deepseq a la conversión a STUArray , los otros algoritmos de clasificación [y conversión, en el caso del vector] toman un poco menos de tiempo, por lo que la diferencia entre el introsort de los hackage.haskell.org/package/vector-algorithms vectoriales y el quicksort de STUArray hace un poco más grande.

list size: 200000 -O2 -fllvm -fllvm-O2
──────── ──────── ──────── ──────── ────────
Data.List.sort 0.663501s 0.665482s 0.652461s 0.792005s
Naive.quicksort 0.587091s 0.577796s 0.585754s 0.667573s
STUArray.quicksort 1.58023s 0.142626s 1.597479s 0.156411s
VSort.vsort 0.820639s 0.139967s 0.888566s 0.143918s

Los tiempos sin optimización son malos para el STUArray . unsafeRead y unsafeWrite deben estar en línea para ser rápidos. Si no está en línea, obtiene una búsqueda de diccionario para cada llamada. Por lo tanto, para el conjunto de datos grande, omito las formas no optimizadas:

list size: 3000000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 16.728576s 16.442377s
Naive.quicksort 14.297534s 12.253071s
STUArray.quicksort 2.307203s 2.200807s
VSort.vsort 2.069749s 1.921943s

Puede ver que una ordenación in situ en una matriz no encajable mutable es mucho más rápida que una ordenación basada en listas si se realiza correctamente. No STUArray si la diferencia entre la clasificación STUArray y la clasificación en el vector mutable sin caja se debe al diferente algoritmo o si los vectores son más rápidos aquí. Como nunca he observado que los vectores sean más rápidos² que los de STUArray , tiendo a creer lo primero. La diferencia entre el quicksort de STUArray y el introsort se debe en parte a la mejor conversión desde y hacia las listas que ofrece el paquete de vector , en parte debido a los diferentes algoritmos.

A sugerencia de Louis Wasserman , he ejecutado un punto de referencia rápido utilizando los otros algoritmos de clasificación del paquete de hackage.haskell.org/package/vector-algorithms , utilizando un conjunto de datos no muy grande. Los resultados no son sorprendentes, los buenos algoritmos de propósito general heapsort, introsort y mergesort funcionan bien, los tiempos se acercan al quicksort en la matriz mutable sin caja (pero, por supuesto, el quicksort se degradaría al comportamiento cuadrático en la entrada casi ordenada, mientras que estos están garantizados O (n * log n) peor caso). Los algoritmos de clasificación de propósito especial AmericanFlag y la clasificación de radix funcionan mal, ya que la entrada no se ajusta bien a su propósito (la clasificación de radix se haría mejor en entradas más grandes con un rango mayor, como es, hace demasiados pases más de los necesarios para los datos). El orden de inserción es, con mucho, el peor, debido a su comportamiento cuadrático.

AmericanFlag:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.083845s 1.084699s
Naive.quicksort 0.981276s 1.05532s
STUArray.quicksort 0.218407s 0.215564s
VSort.vsort 2.566838s 2.618817s
Heap:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.084252s 1.07894s
Naive.quicksort 0.915984s 0.887354s
STUArray.quicksort 0.219786s 0.225748s
VSort.vsort 0.213507s 0.20152s
Insertion:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.168837s 1.066058s
Naive.quicksort 1.081806s 0.879439s
STUArray.quicksort 0.241958s 0.209631s
VSort.vsort 36.21295s 27.564993s
Intro:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.09189s 1.112415s
Naive.quicksort 0.891161s 0.989799s
STUArray.quicksort 0.236596s 0.227348s
VSort.vsort 0.221742s 0.20815s
Merge:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.087929s 1.074926s
Naive.quicksort 0.875477s 1.019984s
STUArray.quicksort 0.215551s 0.221301s
VSort.vsort 0.236661s 0.230287s
Radix:
list size: 300000 -O2 -fllvm-O2
──────── ──────── ────────
Data.List.sort 1.085658s 1.085726s
Naive.quicksort 1.002067s 0.900985s
STUArray.quicksort 0.217371s 0.228973s
VSort.vsort 1.958216s 1.970619s

Conclusión: a menos que tenga una razón específica para no hacerlo, usar uno de los buenos algoritmos de clasificación de propósito general de hackage.haskell.org/package/vector-algorithms , con una envoltura para convertir desde y a listas si es necesario, es la forma recomendada para ordenar listas grandes. (Estos algoritmos también funcionan bien con vectores en caja, en mis mediciones aproximadamente un 50% más lento que sin caja). Para listas cortas, la sobrecarga de la conversión sería tan grande que no se paga.

Ahora, a sugerencia de @ aplicative, un vistazo a los tiempos de clasificación para el introsort de hackage.haskell.org/package/vector-algorithms vectoriales, un quicksort en vectores sin caja y un quicksort mejorado (robando descaradamente la implementación de STUArray unstablePartition ) en STUArray s.

La mejorada STUArray rápida de STUArray :

{-# LANGUAGE BangPatterns #-}
module NQuick (stuqsort) where
import Data.Array.Base (unsafeRead, unsafeWrite, getNumElements)
import Data.Array.ST
import Control.Monad.ST
import Control.Monad (when)
stuqsort :: STUArray s Int Int -> ST s ()
stuqsort arr = do
n <- getNumElements arr
when (n > 1) (myqsort arr 0 (n-1))
myqsort :: STUArray s Int Int -> Int -> Int -> ST s ()
myqsort a lo hi = do
p <- unsafeRead a hi
j <- unstablePartition (< p) lo hi a
h <- unsafeRead a j
unsafeWrite a j p
unsafeWrite a hi h
when (j > lo+1) (myqsort a lo (j-1))
when (j+1 < hi) (myqsort a (j+1) hi)
unstablePartition :: (Int -> Bool) -> Int -> Int -> STUArray s Int Int -> ST s Int
{-# INLINE unstablePartition #-}
unstablePartition f !lf !rg !v = from_left lf rg
where
from_left i j
| i == j = return i
| otherwise = do
x <- unsafeRead v i
if f x
then from_left (i+1) j
else from_right i (j-1)
from_right i j
| i == j = return i
| otherwise = do
x <- unsafeRead v j
if f x
then do
y <- unsafeRead v i
unsafeWrite v i x
unsafeWrite v j y
from_left (i+1) j
else from_right i (j-1)

El vector quicksort:

module VectorQuick (vquicksort) where
import qualified Data.Vector.Unboxed.Mutable as UM
import qualified Data.Vector.Generic.Mutable as GM
import Control.Monad.ST
import Control.Monad (when)
vquicksort :: UM.STVector s Int -> ST s ()
vquicksort uv = do
let li = UM.length uv - 1
ui = UM.unsafeSlice 0 li uv
p <- UM.unsafeRead uv li
j <- GM.unstablePartition (< p) ui
h <- UM.unsafeRead uv j
UM.unsafeWrite uv j p
UM.unsafeWrite uv li h
when (j > 1) (vquicksort (UM.unsafeSlice 0 j uv))
when (j + 1 < li) (vquicksort (UM.unsafeSlice (j+1) (li-j) uv))

El código de tiempo:

{-# LANGUAGE BangPatterns #-}
module Main (main) where
import System.Environment (getArgs)
import System.CPUTime
import System.Random
import Text.Printf
import Data.Array.Unboxed
import Data.Array.ST hiding (unsafeThaw)
import Data.Array.Unsafe (unsafeThaw)
import Data.Array.Base (unsafeAt, unsafeNewArray_, unsafeWrite)
import Control.Monad.ST
import Control.Monad
import qualified Data.Vector.Unboxed as U
import qualified Data.Vector.Unboxed.Mutable as UM
import NQuick
import VectorQuick
import qualified Data.Vector.Algorithms.Intro as I
nextR :: StdGen -> (Int, StdGen)
nextR = randomR (minBound, maxBound)
buildArray :: StdGen -> Int -> UArray Int Int
buildArray sg size = runSTUArray (do
arr <- unsafeNewArray_ (0, size-1)
let fill i g
| i < size = do
let (r, g'') = nextR g
unsafeWrite arr i r
fill (i+1) g''
| otherwise = return arr
fill 0 sg)
buildVector :: StdGen -> Int -> U.Vector Int
buildVector sg size = U.fromList $ take size (randoms sg)
time :: IO a -> IO ()
time action = do
t0 <- getCPUTime
action
t1 <- getCPUTime
let tm :: Double
tm = fromInteger (t1 - t0) * 1e-9
printf "%.3f ms/n" tm
stu :: UArray Int Int -> Int -> IO ()
stu ua sz = do
let !sa = runSTUArray (do
st <- unsafeThaw ua
stuqsort st
return st)
forM_ [0, sz `quot` 2, sz-1] (print . (sa `unsafeAt`))
intro :: U.Vector Int -> Int -> IO ()
intro uv sz = do
let !sv = runST (do
st <- U.unsafeThaw uv
I.sort st
U.unsafeFreeze st)
forM_ [0, sz `quot` 2, sz-1] (print . U.unsafeIndex sv)
vquick :: U.Vector Int -> Int -> IO ()
vquick uv sz = do
let !sv = runST (do
st <- U.unsafeThaw uv
vquicksort st
U.unsafeFreeze st)
forM_ [0, sz `quot` 2, sz-1] (print . U.unsafeIndex sv)
main :: IO ()
main = do
args <- getArgs
let !num = case args of
(a:_) -> read a
_ -> 1000000
!sg <- getStdGen
let !ar = buildArray sg num
!vc = buildVector sg num
!v2 = buildVector sg (foo num)
algos = [ ("Intro", intro v2), ("STUArray", stu ar), ("Vquick", vquick vc) ]
printf "Created data to be sorted, last elements %d %d %d/n" (ar ! (num-1)) (vc U.! (num-1)) (v2 U.! (num-1))
forM_ algos $ /(name, act) -> do
putStrLn name
time (act num)
-- For the prevention of sharing
foo :: Int -> Int
foo n
| n < 0 = -n
| n > 0 = n
| otherwise = 3

Los resultados (solo veces):

$ ./timeSorts 3000000
Intro
587.911 ms
STUArray
402.939 ms
Vquick
414.936 ms
$ ./timeSorts 1000000
Intro
193.970 ms
STUArray
131.980 ms
Vquick
134.979 ms

La STUArray rápida prácticamente idéntica en el STUArray y el vector sin caja toman prácticamente el mismo tiempo, como se esperaba. (La antigua implementación de ordenación rápida fue aproximadamente un 15% más lenta que la introsort. En comparación con los tiempos anteriores, alrededor del 70-75% se gastó la conversión de / a las listas).

En la entrada aleatoria, las acciones rápidas tienen un rendimiento significativamente mejor que el introsort, pero en la entrada casi ordenada, su rendimiento se degradaría, mientras que el introsort no.

¹ Hacer que el código sea polimórfico con STUArray s es un dolor en el mejor de los casos, hacerlo con IOUArray s y tener tanto la clasificación como el envoltorio {-# INLINABLE #-} produce el mismo rendimiento con optimizaciones; sin {-# INLINABLE #-} , el código polimórfico es significativamente más lento.

² Usando los mismos algoritmos, ambos siempre fueron igual de rápidos en la precisión de la medición cuando los comparé (no muy a menudo).

Answer 2

No hay nada mejor que una prueba, ¿verdad? Y los resultados no son sorprendentes: para listas de enteros aleatorios en el rango [0 .. 1000000] ,

list size: 200000 ghc -O2 -fllvm -fllvm-O2
──────── ──────── ──────── ──────── ────────
Data.List.sort 0.878969s 0.883219s 0.878106s 0.888758s
Naïve.quicksort 0.711305s 0.870647s 0.845508s 0.919925s
UArray_IO.quicksort 9.317783s 1.919583s 9.390687s 1.945072s
Vector_Mutable.quicksort 1.48142s 0.823004s 1.526661s 0.806837s

Aquí, Data.List.sort es exactamente lo que es, Naïve.quicksort es el algoritmo que usted cita, UArray_IO.quicksort y Vector_Mutable.quicksort se toman de la pregunta que vinculó a: klapaucius'' y la respuesta de Dan Burton que resulta ser muy En cuanto al rendimiento óptimo, vea qué mejor podría hacerlo Daniel Fischer , ambos envueltos para aceptar las listas (no estoy seguro de haberlo hecho bien):

quicksort :: [Int] -> [Int]
quicksort l = unsafePerformIO $ do
let bounds = (0, length l)
arr <- newListArray bounds l :: IO (IOUArray Int Int)
uncurry (qsort arr) bounds
getElems arr

y

quicksort :: Ord a => [a] -> [a]
quicksort = toList . iqsort . fromList

respectivamente.

Como puede ver, el algoritmo ingenuo no está muy por detrás de la solución mutable con Data.Vector en términos de velocidad para clasificar una lista de enteros generados aleatoriamente, y el IOUArray es en realidad mucho peor . La prueba se llevó a cabo en una computadora portátil Intel i5 con Ubuntu 11.10 x86-64.

Lo siguiente no tiene mucho sentido teniendo en cuenta que, después de todo, las implementaciones mutables están muy por delante de todas las comparadas aquí.

Tenga en cuenta que esto no significa que un buen programa basado en listas siempre pueda mantenerse al día con sus equivalentes implementados de manera que se pueda realizar, pero GHC hace un gran trabajo para acercar el rendimiento. Además, depende, por supuesto, de los datos: estos son los momentos en que las listas generadas aleatoriamente contienen valores entre 0 y 1000 en lugar de 0 y 1000000 como antes, es decir, con muchos duplicados:

list size: 200000 ghc -O2 -fllvm -fllvm-O2
──────── ──────── ──────── ──────── ────────
Data.List.sort 0.864176s 0.882574s 0.850807s 0.857957s
Naïve.quicksort 1.475362s 1.526076s 1.475557s 1.456759s
UArray_IO.quicksort 24.405938s 5.255001s 23.561911s 5.207535s
Vector_Mutable.quicksort 3.449168s 1.125788s 3.202925s 1.117741s

Por no hablar de matrices pre-ordenadas.

Lo que es bastante interesante, (solo se hace evidente con tamaños realmente grandes, que requieren rtsopts para aumentar la capacidad de pila), es cómo ambas implementaciones mutables se vuelven significativamente más lentas con -fllvm -O2 :

list size: 3⋅10⁶ ghc -O1 -fllvm-O1 -O2 -fllvm-O2
──────── ──────── ──────── ──────── ────────
Data.List.sort 23.897897s 24.138117s 23.708218s 23.631968s
Naïve.quicksort 17.068644s 19.547817s 17.640389s 18.113622s
UArray_IO.quicksort 35.634132s 38.348955s 37.177606s 49.190503s
Vector_Mutable.quicksort 17.286982s 17.251068s 17.361247s 36.840698s

Me parece lógico que a las implementaciones inmutables les vaya mejor en llvm (¿no hace todo de manera inmutable en algún nivel?), Aunque no entiendo por qué esto solo se hace evidente como una ralentización de las versiones mutables con una optimización alta. y grandes tamaños de datos.

Programa de pruebas:

$ cat QSortPerform.hs
module Main where
import qualified Data.List(sort)
import qualified Naïve
import qualified UArray_IO
import qualified Vector_Mutable
import Control.Monad
import System.Random
import System.Environment
sortAlgos :: [ (String, [Int]->[Int]) ]
sortAlgos = [ ("Data.List.sort", Data.List.sort)
, ("Naïve.quicksort", Naïve.quicksort)
, ("UArray_IO.quicksort", UArray_IO.quicksort)
, ("Vector_Mutable.quicksort", Vector_Mutable.quicksort) ]
main = do
args <- getArgs
when (length args /= 2) $ error "Need 2 arguments"
let simSize = read $ args!!1
randArray <- fmap (take simSize . randomRs(0,1000000)) getStdGen
let sorted = case filter ((== args!!0) . fst) sortAlgos of
[(_, algo)] -> algo randArray
_ -> error $ "Argument must be one of "
++ show (map fst sortAlgos)
putStr "First element: "; print $ sorted!!0
putStr "Middle element: "; print $ sorted!!(simSize`div`2)
putStr "Last element: "; print $ sorted!!(simSize-1)

que toma el nombre del algoritmo y el tamaño de la matriz en la línea de comandos. La comparación de tiempo de ejecución se realizó con este programa:

$ cat PerformCompare.hs
module Main where
import System.Process
import System.Exit
import System.Environment
import Data.Time.Clock
import Data.List
import Control.Monad
import Text.PrettyPrint.Boxes
compiler = "ghc"
testProgram = "./QSortPerform"
flagOpts = [[], ["-O2"], ["-fllvm"], ["-fllvm","-O2"]]
algos = ["Data.List.sort","Naïve.quicksort","UArray_IO.quicksort","Vector_Mutable.quicksort"]
main = do
args <- getArgs
let testSize = case args of
[numb] -> read numb
_ -> 200000
results <- forM flagOpts $ /flags -> do
compilerExitC <- verboseSystem
compiler $ testProgram : "-fforce-recomp" : flags
when (compilerExitC /= ExitSuccess) .
error $ "Compiler error /"" ++ show compilerExitC ++"/""
algoCompare <- forM algos $ /algo -> do
startTime <- getCurrentTime
exitC <- verboseSystem testProgram [algo, show testSize]
endTime <- getCurrentTime
when (exitC /= ExitSuccess) .
error $ "Program error /"" ++ show exitC ++"/""
return . text . show $ diffUTCTime endTime startTime
return . vcat right $ text(concat flags)
: text("────────")
: algoCompare
let table = hsep 2 bottom
$ vcat left (map text $ ("list size: "++show testSize)
: "────────"
: algos )
: results
printBox table
verboseSystem :: String -> [String] -> IO ExitCode
verboseSystem cmd args = do
putStrLn . unwords $ cmd : args
rawSystem cmd args