使用向量在Haskell中提高性能

Pot*_*ato 23 algorithm haskell

我对Haskell很新,我对使用不纯(可变)数据结构可以提高性能有疑问.我正在尝试拼凑一些我听过的不同的东西,所以如果我的术语不完全正确,或者如果有一些小错误,请耐心等待.

为了具体化,请考虑快速排序算法(取自Haskell wiki).

quicksort :: Ord a => [a] -> [a]
quicksort []     = []
quicksort (p:xs) = (quicksort lesser) ++ [p] ++ (quicksort greater)
    where
        lesser  = filter (< p) xs
        greater = filter (>= p) xs
Run Code Online (Sandbox Code Playgroud)

这不是"真正的快速入门"."真正的"快速排序算法就位,但事实并非如此.这是非常低效的内存.

另一方面,可以在Haskell中使用向量来实现就地快速排序.这个stackoverflow答案给出了一个例子.

第二种算法比第一种算法快多少?Big O符号在这里没有用,因为性能的提高将来自更有效地使用内存,而不是更好的算法(对吧?).我厌倦了自己构建一些测试用例,但是我很难让事情运行起来.

理想的答案可以让我们了解在理论上使原位Haskell算法更快的原因,以及某些测试数据集上运行时间的示例比较.

Dan*_*her 21

另一方面,可以在Haskell中使用向量来实现就地快速排序.

第二种算法比第一种算法快多少?

当然,这取决于实施.正如下面可以看到,对于不能太短列表,一个体面就地排序上一个可变的载体或阵列比排序列表快得多,即使被包括用于从和到列表转换的时间(和转换构成大部分时间).

但是,列表算法会产生增量输出,而数组/矢量算法在完成之前不会产生任何结果,因此排序列表仍然是可取的.

我不确切知道链接的可变数组/向量算法做错了什么.但是他们做错了.

对于可变向量代码,似乎它使用了盒装向量,并且它是多态的,两者都可以产生显着的性能影响,尽管如果函数是多态,则无关紧要{-# INLINABLE #-}.

对于IOUArray代码来说,它看起来很有趣,但很慢.它使用a IORef,readArray并且writeArray没有明显的严格性.那么糟糕的时间并不太令人惊讶.

使用一个更加直接的(单态)C代码转换,使用STUArray包装器使其在列表¹上工作,

{-# 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 ()
Run Code Online (Sandbox Code Playgroud)

和一个好的排序包装(Introsort,而不是快速排序)在未装箱的矢量上,

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)
Run Code Online (Sandbox Code Playgroud)

我得到的时间更符合预期(注意:对于这些时间,deepseq在调用排序算法之前已经编辑了随机列表.没有它,转换为a STUArray会慢得多,因为它会首先评估一长串的thunk确定长度.矢量包的fromList转换不会遇到这个问题.转移到转换,其他排序[和转换,在矢量情况下]算法花费的时间少一点,所以矢量之间的差异-algorithms的introsort和quicksort 变得有点大.):deepseqSTUArraySTUArray

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
Run Code Online (Sandbox Code Playgroud)

没有优化的时代预计会对此不利STUArray.unsafeRead并且unsafeWrite必须内联才能快速.如果没有内联,则会为每个调用获取字典查找.因此,对于大型数据集,我省略了未经优化的方法:

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
Run Code Online (Sandbox Code Playgroud)

您可以看到,如果正确完成,可变无盒装数组上的就地排序比基于列表的排序要快得多.STUArray未装箱的可变矢量的排序和排序之间的差异是由于不同的算法还是矢量确实在这里更快,我不知道.由于我从未观察过矢量比STUArrays 更快,我倾向于相信前者.STUArray快速排序和内部排序 之间的差异部分是由于向量包提供的更好的转换和转换到列表,部分原因在于不同的算法.


在Louis Wasserman的建议下,我使用了一个不太大的数据集,使用vector-algorithms包中的其他排序算法运行了一个快速基准测试.结果并不奇怪,好的通用算法heapsort,introsort和mergesort都做得很好,接近未装箱的可变阵列上的快速排序(但是当然,快速排序会在几乎排序的输入上降级为二次行为,而这些保证O(n*log n)最坏的情况).特殊目的的排序算法AmericanFlag和基数排序很糟糕,因为输入不能很好地适应它们的目的(基数排序在更大范围的输入上会做得更好,因为它的传递次数比需要的多得多.数据).由于其二次行为,插入排序是最差的.

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
Run Code Online (Sandbox Code Playgroud)

结论:除非您有特定的理由不使用矢量算法中的一个良好的通用排序算法,并且必要时使用包装器从列表转换到列表,否则是对大型列表进行排序的推荐方法.(这些算法也适用于盒装向量,在我的测量中比未装箱大约慢50%.)对于短列表,转换的开销会非常大,以至于无法支付.


现在,在@ applicative的建议下,看一下矢量算法的内部分类的排序时间,无盒式向量的快速排序和unstablePartition对STUArrays 的改进(无耻地窃取实现)快速排序.

改进的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)
Run Code Online (Sandbox Code Playgroud)

该载体快速排序:

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))
Run Code Online (Sandbox Code Playgroud)

时间码:

{-# 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
Run Code Online (Sandbox Code Playgroud)

结果(仅限次数):

$ ./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
Run Code Online (Sandbox Code Playgroud)

STUArray正如预期的那样,实际上相同的快速排序和未装箱的矢量几乎同时进行.(旧的quicksort实现比introsort慢了大约15%.与上面的时间相比,大约有70-75%用于从/向列表转换.)

在随机输入上,快速排序的性能明显优于introsort,但在几乎排序的输入上,它们的性能会降低,而内部排序则不会.


¹使用STUArrays 使代码具有多态性是最痛苦的,使用IOUArrays并使排序和包装器{-# INLINABLE #-}通过优化产生相同的性能 - 没有,多态代码明显变慢.

²使用相同的算法,当我进行比较(不常见)时,两者在测量精度范围内总是同样快.


lef*_*out 17

没有比测试更好的了,对吧?结果并不令人惊讶:对于范围内的随机整数列表[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
Run Code Online (Sandbox Code Playgroud)

在这里,Data.List.sort就是它是什么,Naïve.quicksort是你引用的算法,UArray_IO.quicksort并Vector_Mutable.quicksort从你所链接的问题中得到:klapaucius'和Dan Burton的答案 ,结果是非常不理想的表现,看看Daniel Fischer可以做得更好,两者都被包裹以便接受列表(不确定我是否完全正确):

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
Run Code Online (Sandbox Code Playgroud)

和

quicksort :: Ord a => [a] -> [a]
quicksort = toList . iqsort . fromList
Run Code Online (Sandbox Code Playgroud)

分别.

正如您所看到的,在Data.Vector对随机生成的整数列表进行排序的速度方面,天真算法与可变解决方案相差不远,IOUArray实际上更糟糕.测试是在运行Ubuntu 11.10 x86-64的Intel i5笔记本电脑上进行的.


考虑到ɢᴏᴏᴅ可变实现毕竟仍远远超过这里比较的所有实现,因此以下并没有多大意义.

请注意,这并不意味着一个好的基于列表的程序可以始终跟上其可变实现的等价物,但GHC肯定能够很好地实现性能接近.此外,它当然取决于数据:这些是随机生成的排序列表包含0到1000之间的值而不是如上所述的0和1000000的时间,即有许多重复:

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
Run Code Online (Sandbox Code Playgroud)

更不用说预先排序的数组了.

什么是挺有意思的,(只变成真正具有大尺寸,这需要rtsopts增加堆栈容量明显),既是可变的实施如何成为显著慢有-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
Run Code Online (Sandbox Code Playgroud)

对我来说,不可变的实现在llvm上表现得更好(在某种程度上它不能在某种程度上做到不可变吗?)似乎有点合乎逻辑(虽然我不明白为什么这只会在高优化时变得明显变慢为可变版本)和大数据量.


测试程序:

$ 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)
Run Code Online (Sandbox Code Playgroud)

它在命令行中获取算法名称和数组大小.使用此程序进行运行时比较:

$ 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
Run Code Online (Sandbox Code Playgroud)