Warm tip: This article is reproduced from serverfault.com, please click

algorithm-如何优化我的 Haskell 以免内存不足

(algorithm - How can I optimise my Haskell so I don't run out of memory)

发布于 2021-03-06 17:16:50

对于在线算法课程,我正在尝试编写一个程序,该程序使用近似算法计算城市的旅行商距离:

  1. 在第一个城市开始游览。
  2. 反复访问旅行团尚未访问的最近的城市。如果出现平局,请前往最近的指数最低的城市。例如,如果第三个和第五个城市与第一个城市的距离相同(并且比任何其他城市都近),那么游览应该从第一个城市到第三个城市开始。
  3. 一旦每个城市都被访问了一次,就返回第一个城市完成旅行。

我正在尝试用 Haskell 编写一个解决方案,我让它处理小数据集,但它在大输入时内存不足(该课程有 ~33000 个城市的输入)

-- Fold data: cities map, distances map, visited map, list of visited cities and each distance,
-- and current city
data TS = TS (M.Map Int City) (M.Map (Int,Int) Double) (M.Map Int Bool) ([(Int,Double)]) (Int)

run :: String -> String
run input = let cm = parseInput input -- cityMap contains cities (index,xPos,yPos)
                n = length $ M.keys cm
                dm = buildDistMap cm -- distanceMap :: M.Map (Int,Int) Double
                                     -- which is the distance between cities a and b
                ts = TS cm dm (M.fromList [(1,True)]) [(1,0.0)] 1
                (TS _ _ _ beforeLast _) = foldl' (\ts _ -> exec ts n) ts [2..n]
                completed = end beforeLast dm
             in show $ floor $ sum $ map (\(_,d) -> d) $ completed

exec :: TS -> Int -> TS
exec (TS cm dm visited ordered curr) n =
  let candidateIndexes = [(i)|i<-[1..n],M.member i visited == False]
      candidates = map (\i -> let (Just x) = M.lookup (curr,i) dm in (x,i)) candidateIndexes
      (dist,best) = head $ sortBy bestCity candidates
      visited' = M.insert best True visited
      ordered' = (best,dist) : ordered
   in  TS cm dm visited' ordered' best

end :: [(Int,Double)] -> M.Map (Int,Int) Double -> [(Int,Double)]
end ordering dm = let (latest,_) = head ordering
                      (Just dist) = M.lookup (latest,1) dm
                   in (1,dist) : ordering

bestCity :: (Double,Int) -> (Double,Int) -> Ordering
bestCity (d1,i1) (d2,i2) =
  if compare d1 d2 == EQ
     then compare i1 i2
     else compare d1 d2

起初,我将该函数编写exec为递归函数,而不是通过foldl'. 我认为将其更改为使用foldl'foldl'严格解决我的问题然而,它似乎对内存使用没有影响。我试过不使用优化和-O2优化来编译我的程序

我知道它必须以某种方式在内存中保留多个循环,因为我可以对 34000 个数字进行排序而不会出现问题

> sort $ [34000,33999..1]

我到底做错了什么?

以防万一这里有任何用处是parseInputbuildDistMap功能,但它们不是我问题的根源

data City = City Int Double Double deriving (Show, Eq)

-- Init
parseInput :: String -> M.Map Int City
parseInput input =
  M.fromList
  $ zip [1..]
  $ map ((\(i:x:y:_) -> City (read i) (read x) (read y)) . words)
  $ tail
  $ lines input

buildDistMap :: M.Map Int City -> M.Map (Int,Int) Double
buildDistMap cm =
  let n = length $ M.keys cm
      bm = M.fromList $ zip [(i,i)|i<-[1..n]] (repeat 0) :: M.Map (Int,Int) Double
      perms = [(x,y)|x<-[1..n],y<-[1..n],x/=y]
   in foldl' (\dm (x,y) -> M.insert (x,y) (getDist cm dm (x,y)) dm) bm perms

getDist :: M.Map Int City -> M.Map (Int,Int) Double -> (Int,Int) -> Double
getDist cm dm (x,y) =
  case M.lookup (y,x) dm
        of (Just v) -> v
           Nothing -> let (Just (City _ x1 y1)) = M.lookup x cm
                          (Just (City _ x2 y2)) = M.lookup y cm
                       in eDist (x1,y1) (x2,y2)

eDist :: (Double,Double) -> (Double,Double) -> Double
eDist (x1,y1) (x2,y2) = sqrt $ p2 (x2 - x1) + p2 (y2 - y1)
  where p2 x = x ^ 2

和一些测试输入

tc1 = "6\n\
  \1 2 1\n\
  \2 4 0\n\
  \3 2 0\n\
  \4 0 0\n\
  \5 4 3\n\
  \6 0 3"
Questioner
mattematt
Viewed
0
Carl 2021-03-07 05:58:58
data TS = TS (M.Map Int City) (M.Map (Int,Int) Double) (M.Map Int Bool) ([(Int,Double)]) (Int)



(TS _ _ _ beforeLast _) = foldl' (\ts _ -> exec ts n) ts [2..n]



exec :: TS -> Int -> TS
exec (TS cm dm visited ordered curr) n =
  let ...
  in  TS cm dm visited' ordered' best

foldl'比你希望的要少得多。它使TS构造函数的每一步进行评估,但没有在评估过程中需要visited'ordered'或者best进行评估。cm并且dm不会在循环中修改,因此它们无法叠加未评估的 thunk。)

解决的最好办法是使评价TS由返回构造exec取决于评估visited'ordered'以及best足够。

M.Map始终是脊椎严格的,因此评估 map意味着评估整个结构。这些值是否也取决于你如何导入它,但事实证明这与此处无关。你插入的值是一个空构造函数,因此它已经被完全评估。所以评估visited'到 WHNF 就足够了。

Int不是嵌套类型,因此评估bestWHNF 就足够了。

[(Int, Double)](外部括号是多余的,列表括号对其内容进行分组)有点棘手。列表不是脊椎严格的,对也不是严格的。但是从构造模式来看,这是一个 prepend-only 结构。因此,你无需担心尾巴。如果列表被评估进来,只要新的头部存在,输出就会被评估。不幸的是,这意味着你必须小心对待这对。它的一半与best上面构造的值相同,所以还不错。如果它被评估,它被评估!(尽管这确实表明你不需要在每次迭代时都传递它,你可以只使用 的前面ordered。)该对的另一半是 a Double,它也是非嵌套的,因此 WHNF 就足够了。

在这种特殊情况下,由于需要不同的方法,我可能只是使用seq.

let ... all the same stuff ...
in  visited' `seq` dist `seq` best `seq` TS ... all the same stuff ...

请注意,我很小心地强制使用最少数量的值来删除不必要的 thunk 嵌套。(,)(:)不需要构造也进行评估,只有他们的观点-在嵌套的thunk可能积聚的地方。<thunk <expression> <expression>>之间的内存消耗有什么区别<constructor <expression> <expression>>?)