计算大型列表的最小/最大/总和时的 Haskell 性能

Question

我一直在试验以下 Haskell 代码：

data Foo = Foo
  { fooMin :: Float
  , fooMax :: Float
  , fooSum :: Float
  } deriving Show


getLocalFoo :: [Float] -> Foo
getLocalFoo x = Foo a b c
  where
    a = minimum x
    b = maximum x
    c = sum x

getGlobalFoo :: [Foo] -> Foo
getGlobalFoo x = Foo a b c 
  where
    a = minimum $ fmap fooMin x
    b = maximum $ fmap fooMax x
    c = sum $ fmap fooSum x


main :: IO()
main = do
  let numItems = 2000
  let numLists = 100000
  putStrLn $ "numItems: " ++ show numItems
  putStrLn $ "numLists: " ++ show numLists

  -- Create an infinite list of lists of floats, x is [[Float]]
  let x = take numLists $ repeat [1.0 .. numItems] 

  -- Print two first elements of each item
  print $ take 2 (map (take 2) x)

  -- First calculate local min/max/sum for each float list 
  -- then calculate the global min/max/sum based on the results.
  print . getGlobalFoo $ fmap getLocalFoo x

并在调整 numItems 和 numLists 时依次测试运行时：

小尺寸：

numItems: 4.0
numLists: 2
[[1.0,2.0],[1.0,2.0]]
Foo {fooMin = 1.0, fooMax = 4.0, fooSum = 20.0}

real    0m0.005s
user    0m0.004s
sys 0m0.001s

大尺寸：

numItems: 2000.0
numLists: 100000
[[1.0,2.0],[1.0,2.0]]
Foo {fooMin = 1.0, fooMax = 2000.0, fooSum = 1.9999036e11}

real    0m33.116s
user    0m33.005s
sys 0m0.109s

我在没有考虑性能的情况下以我认为直观和幼稚的方式编写了此代码，但是我担心这远非最佳代码，因为我实际上可能会以比必要的方式更多次地折叠列表？

谁能建议更好地实施这个测试？

Answer 1

使用foldl 库在一次通过中有效地运行多个折叠。事实上，它非常擅长这一点，您无需将列表拆分为子列表。您可以将所有列表连接在一起形成一个巨大的列表并直接折叠。

方法如下：

import Control.Applicative
import qualified Control.Foldl as L

data Foo = Foo
  { fooMin :: Maybe Float
  , fooMax :: Maybe Float
  , fooSum :: Float
  } deriving Show

foldFloats :: L.Fold Float Foo
foldFloats = Foo <$> L.minimum <*> L.maximum <*> L.sum
-- or: foldFloats = liftA3 Foo L.minimum L.maximum L.sum

main :: IO()
main = do
    let numItems = 2000
    let numLists = 100000
    putStrLn $ "numItems: " ++ show numItems
    putStrLn $ "numLists: " ++ show numLists

    -- Create an infinite list of lists of floats, x is [[Float]]
    let x = replicate numLists [1.0 .. numItems] 

    -- Print two first elements of each item
    print $ take 2 (map (take 2) x)

    print $ L.fold foldFloats (concat x)

与您的代码的主要区别是：

• 我用的是replicate n，和take n . repeat是一样的。其实replicate就是这样定义的

• 我不会单独处理子列表。我只是 concat 将它们全部放在一起，然后一次性折叠。

• 我使用Maybe 作为最小值和最大值，因为我需要处理空列表的情况。

• 这段代码更快

这里是数字：

$ time ./fold
numItems: 2000.0
numLists: 100000
[[1.0,2.0],[1.0,2.0]]
Foo {fooMin = Just 1.0, fooMax = Just 2000.0, fooSum = 3.435974e10}

real  0m5.796s
user  0m5.756s
sys   0m0.024s

foldl 是一个非常小巧且易于学习的库。你可以通过here了解更多。

Answer 2

Monoids 来拯救。你所有的操作——总和、最小值和最大值——都可以表示为幺半群。对于最小值和最大值，我们需要将其从semigroups 包装到Option 中，因为我们需要以某种方式表示空集合的最小值和最大值。 （另一种方法是将我们自己限制在非空集合中，然后我们可以使用半群而不是幺半群。）

我们需要做的另一件事是确保在每个步骤中强制执行所有计算。为此，我们声明 Foo 的 NFData 实例，添加一些我们使用的 monoid 类型的缺失实例，以及在折叠操作期间强制值的辅助函数。

import Control.DeepSeq
import qualified Data.Foldable as F
import Data.Semigroup

-- Declare the data type so that each field is a monoid.
data Foo a = Foo
  { fooMin :: Option (Min a)
  , fooMax :: Option (Max a)
  , fooSum :: Sum a
  } deriving Show

-- Make a Monoid instance - just by combining individual fields.
instance (Ord a, Num a) => Monoid (Foo a) where
  mempty = Foo mempty mempty mempty
  mappend (Foo n1 x1 s1) (Foo n2 x2 s2) = Foo (n1 <> n2) (x1 <> x2) (s1 <> s2)

-- Add missing NFData instances
instance (NFData a) => NFData (Option a) where
  rnf (Option x) = rnf x `seq` ()
instance (NFData a) => NFData (Min a) where
  rnf (Min x) = rnf x `seq` ()
instance (NFData a) => NFData (Max a) where
  rnf (Max x) = rnf x `seq` ()
instance (NFData a) => NFData (Sum a) where
  rnf (Sum x) = rnf x `seq` ()

-- Also add an instance for Foo
instance (NFData a) => NFData (Foo a) where
  rnf (Foo n x s) = rnf n `seq` rnf x `seq` rnf s `seq` ()

-- Convert a single element into Foo.
locFoo :: a -> Foo a
locFoo x = Foo (return $ Min x) (return $ Max x) (Sum x)

-- A variant of foldMap that uses left fold and forces monoid
-- elements on the way.
foldMap' :: (F.Foldable f, Monoid m, NFData m) => (a -> m) -> f a -> m
foldMap' f = F.foldl' (\m x -> (mappend $!! m) (f x)) mempty

main :: IO()
main = do
  let numItems = 2000
  let numLists = 100000
  putStrLn $ "numItems: " ++ show numItems
  putStrLn $ "numLists: " ++ show numLists

  -- Create an infinite list of lists of floats, x is [[Float]]
  let x = take numLists $ repeat [1.0 .. numItems]  :: [[Float]]

  -- Print two first elements of each item
  print $ take 2 (map (take 2) x)

  -- First calculate local min/max/sum for each float list 
  -- then calculate the global min/max/sum based on the results.
  print . foldMap' (foldMap' locFoo) $ x

Answer 3

也许单折更便宜。尝试运行一些测试，例如：

{-# LANGUAGE BangPatterns #-}
import Data.List

getLocalFoo :: [Float] -> Foo
getLocalFoo [] = error "getLocalFoo: empty list"
getLocalFoo (x:xs) = foldl' f (Foo x x x) xs
  where f (Foo !min1 !max1 !sum1) y =
          Foo (min1 `min` y) (max1 `max` y) (sum1 + y)

和getGlobalFoo类似。