Использование векторов для повышения производительности в Haskell

Ответ 1

Нет ничего лучше теста, не так ли? И результаты не неудивительны: для списков случайных целых чисел в диапазоне [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

Здесь Data.List.sort это именно то, что есть, Naïve.quicksort - это алгоритм, который вы цитировали, UArray_IO.quicksort и Vector_Mutable.quicksort взяты из вопроса, с которым вы связались: klapaucius ' и Дэн Бертон ответ , которые оказываются очень субоптимальными по производительности, см., что лучше Даниэль Фишер мог это сделать, оба были обернуты так, чтобы принимать списки (не уверен, что я понял это совершенно правильно):

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

и

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

соответственно.

Как вы можете видеть, наивный алгоритм не намного отстает от изменяемого решения с Data.Vector с точки зрения скорости сортировки списка целых чисел, генерируемых случайным образом, а IOUArray на самом деле намного хуже. Тест проводился на ноутбуке Intel i5 под управлением Ubuntu 11.10 x86-64.

Обратите внимание, что это не означает, что хорошая программа на основе списков всегда может идти в ногу с ее изменчиво реализованными эквивалентами, но 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

Не говорить о предварительно отсортированных массивах.

Что довольно интересно, (становится очевидным только с действительно большими размерами, для которых требуется 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

Кажется логичным для меня, что неизменные реализации лучше работают на 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)

который принимает имя и размер алгоритма в командной строке. Сравнение времени выполнения с этой программой:

$ 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

Ответ 2

С другой стороны, в Haskell можно использовать векторы для реализации быстрой сортировки на месте.

Насколько быстрее второй алгоритм, чем первый?

Это зависит от реализации, конечно. Как можно видеть ниже, для не слишком коротких списков достойный внешний вид на изменяемом векторе или массиве намного быстрее, чем сортировка списков, даже если включено время для преобразования из списка и в списки (и это преобразование составляет основная часть времени).

Однако алгоритмы списка производят инкрементный вывод, в то время как алгоритмы массива/вектора не дают результата до их завершения, поэтому списки сортировки могут быть предпочтительнее.

Я не знаю точно, как сработали измененные алгоритмы массива/вектора. Но они сделали что-то не так.

Для изменяемого векторного кода кажется, что он использовал вставные векторы, и он был полиморфным, оба могут иметь значительное влияние на производительность, хотя полиморфизм не должен иметь значения, если функции {-# INLINABLE #-}.

Для кода IOUArray, это выглядит забавно, но медленно. Он использует IORef, readArray и writeArray и не имеет очевидной строгости. Тогда ужасные времена не слишком удивительны.

Используя более прямой перевод (мономорфного) кода C с помощью STUArray с оберткой, чтобы заставить ее работать над списками1,

{-# 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 ()

и обертка вокруг хорошего сорта (Introsort, а не quicksort) на unboxed векторах,

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)

Я получаю время больше в соответствии с ожиданиями ( Примечание:. Для этих таймингов случайный список был deepseq ed до вызова алгоритма сортировки. Без этого преобразование в STUArray будет намного медленнее, так как сначала будет оценивать длинный список трюков для определения длины. Преобразование fromList vector пакет не страдает от этой проблемы. Перемещая deepseq к преобразованию в STUArray, другие алгоритмы сортировки [и преобразования в векторном случае] занимают немного меньше времени, поэтому разница между vector-algorithms 'introsort, а quicksort STUArray будет немного больше.):

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

Время без оптимизации, как ожидается, плохо для 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

Вы можете видеть, что сортировка inplace в изменяемом unboxed массиве намного быстрее, чем сортировка по списку, если все сделано правильно. Независимо от того, отличается ли разница между сортировкой STUArray и сортировкой в ​​unboxed изменчивом векторе из-за разного алгоритма или действительно ли векторы здесь быстрее, я не знаю. Поскольку я никогда не наблюдал, чтобы векторы были быстрее ², чем STUArray s, я склонен считать, что первый. Разница между быстрой сортировкой STUArray и introsort частично объясняется лучшим преобразованием из и в списки, которые предлагает пакет vector, частично из-за разных алгоритмов.

В предложение Луи Вассермана, я провел быстрый тест с использованием других алгоритмов сортировки из vector-algorithms, используя не слишком большой набор данных. Результаты не удивительны, хорошие алгоритмы универсального алгоритма heapsort, introsort и mergesort все хорошо, временами рядом с quicksort в unboxed изменяемом массиве (но, конечно, quicksort будет деградировать до квадратичного поведения на почти отсортированном входе, тогда как эти гарантированы O (n * log n) наихудший случай). Алгоритмы сортировки специального назначения AmericanFlag и сортировка по методу radix делают плохо, так как вход не подходит для их цели (сортировка ради лучше будет работать на больших входах с большим диапазоном, так как это делает слишком много больше проходов, чем необходимых для данных). Сортировка вставки, безусловно, наихудшая из-за ее квадратичного поведения.

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

Заключение: если у вас нет конкретной причины, не используя один из хороших алгоритмов сортировки общего назначения из vector-algorithms, рекомендуется обертка для конвертирования из списка и в списки, если это необходимо, - это рекомендуемый способ сортировки больших списков. (Эти алгоритмы также хорошо работают с коробочными векторами, в моих измерениях примерно на 50% медленнее, чем unboxed.) Для коротких списков накладные расходы на преобразование были бы настолько большими, чтобы они не платили.

Теперь, в @applicative suggestion, посмотрите время сортировки vector-algorithms 'introsort, quicksort на unboxed vectors и (бесстыдно крадут реализацию unstablePartition) quicksort на STUArray s.

Улучшенная STUArray quicksort:

{-# 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)

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))

Временной код:

{-# 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

Результаты (только время):

$ ./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

Практически идентичные quicksorts на STUArray и unboxed vector принимают практически одно и то же время, как и ожидалось. (Старая реализация quicksort была примерно на 15% медленнее, чем интросорт. По сравнению с вышеприведенными временами, около 70-75% было потрачено на преобразование из/в списки.)

На случайном входе операторы быстрой сортировки выполняют значительно лучше, чем интросорт, но на почти отсортированном входе их производительность будет ухудшаться, а introsort не будет.

¹ Создание полиморфизма кода с помощью STUArray является в лучшем случае болью, делая его с помощью IOUArray и имея как сортировку, так и обертку {-# INLINABLE #-} дает такую ​​же производительность с оптимизацией - без, полиморфный код значительно медленнее.

² Используя те же алгоритмы, оба всегда были одинаково быстрыми в пределах точности измерения, когда я сравнивал (не очень часто).