如何优化我的 Haskell 以免 运行 内存不足
How can I optimise my Haskell so I don't run out of memory
对于在线算法课程,我正在尝试编写一个程序,使用近似算法计算城市的旅行商距离:
- 从第一个城市开始游览。
- 重复访问最近的旅行团尚未访问的城市。如果出现平局,则前往距离最低的最近城市
指数。例如,如果第三和第五城市都有
与第一个城市的距离相同(并且比任何其他城市都近
城市),那么旅游应该从第一个城市开始
第三个城市。
- 一旦每个城市都被访问过一次,return到第一个城市完成游览。
我正在尝试在 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]
我到底做错了什么?
parseInput
和 buildDistMap
函数以防万一,但它们不是我问题的根源
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\
2 1\n\
4 0\n\
2 0\n\
0 0\n\
4 3\n\
0 3"
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。)
解决这个问题的最佳方法是让 exec
返回的 TS
构造函数的计算依赖于对 visited'
、ordered'
和 [=16 的计算=] 足够了。
M.Map
始终是 spine-strict,因此评估地图意味着评估整个结构。这些值是否也取决于您如何导入它,但事实证明这与此处无关。您要插入的值是空构造函数,因此它已被完全计算。因此,将 visited'
评估为 WHNF 就足够了。
Int
不是嵌套类型,因此将 best
评估为 WHNF 就足够了。
[(Int, Double)]
(外部 parens 是多余的,列表括号对其内容进行分组)有点棘手。列表不是 spine 严格的,pairs 也不是严格的。但是看构造模式,这是一个prepend-only结构。因此,您无需担心尾巴。如果列表被评估进来,只要新的头是,输出就会被评估。不幸的是,这意味着你必须对这对有点小心。它的一半与上面构造的 best
值相同,所以还不错。如果它被评估,它被评估! (尽管这确实表明您不需要在每次迭代时都传递它,您可以只使用 ordered
的前面。)该对的另一半是 Double
,它也是 non-nested,所以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>>
在内存消耗上有什么区别?)
感谢 Carl 非常详细的回答。还要感谢 Daniel 指出缓存大量距离实际上可能会导致我的记忆问题。我假设因为我的代码已经通过了那个函数,所以我有足够的内存来实现它——忘记 Haskell 是懒惰的,只是在我实际使用它时在 exec
函数中构建那个映射。
我现在以更简洁的方式解决了这个问题。我正在使用我仍然需要访问的所有城市索引的 Data.Set
,然后因为城市是按照 X
值的顺序给出的,我知道飞机上最近的城市也是最近的城市指数。知道这一点后,我设置了一个值,以便在每次迭代时从索引的任一侧获取一个切片,并使用该切片检查到我当前城市的距离,这使我可以在每次迭代时计算到下一个城市的距离,而无需缓存大量数据。
-- How many cities in each direction (index) to consider
-- smaller is faster but less accurate
searchWidth = 1000 :: Int
data TS = TS (M.Map Int City) (S.Set Int) [(Double,Int)] Int
run :: String -> String
run input =
let cm = parseInput input
n = length $ M.keys cm
toVisit = S.fromList [1..n]
ts = TS cm toVisit [(0.0,1)] 1
(TS _ _ beforeLast _) = foldl' (\ts i -> trace (concat [show i,"/",show n]) exec ts) ts [2..n]
afterLast = end cm beforeLast
in show $ floor $ sum $ map (\(d,_) -> d) afterLast
exec :: TS -> TS
exec (TS cm toVisit visited curr) =
let (Just (City _ cx cy)) = M.lookup curr cm
index = S.findIndex curr toVisit
toVisit' = S.deleteAt index toVisit
lb = let x = index - searchWidth in if x < 0 then 0 else x
ub = let x = index + searchWidth - lb in if x >= length toVisit' then (length toVisit') else x
candidateIndexes = S.take ub $ S.drop lb toVisit'
candidates = S.map (\i -> let (Just (City _ x y)) = M.lookup i cm in (eDist (x,y) (cx,cy),i)) candidateIndexes
(dist,next) = S.findMin candidates
visited' = (dist,next) : visited
in toVisit' `seq` dist `seq` next `seq` TS cm toVisit' visited' next
end :: M.Map Int City -> [(Double,Int)] -> [(Double,Int)]
end cm visited =
let (_,currI) = head visited
(Just (City _ cx cy)) = M.lookup currI cm
(Just (City _ lx ly)) = M.lookup 1 cm
dist = eDist (cx,cy) (lx,ly)
in (dist,1) : visited
使用 Data.Set
还带来了额外的好处,它会自动对内部的值进行排序,让获得下一个旅行地点变得微不足道。
我意识到这不是世界上最好的 Haskell 代码,我正在做一些顽皮的事情,比如直接从地图查找中匹配 Just
而不是使用 [=16] =] 值。此外,有人向我指出我应该使用记录而不是 data
类型来构造我的 TS
对于在线算法课程,我正在尝试编写一个程序,使用近似算法计算城市的旅行商距离:
- 从第一个城市开始游览。
- 重复访问最近的旅行团尚未访问的城市。如果出现平局,则前往距离最低的最近城市 指数。例如,如果第三和第五城市都有 与第一个城市的距离相同(并且比任何其他城市都近 城市),那么旅游应该从第一个城市开始 第三个城市。
- 一旦每个城市都被访问过一次,return到第一个城市完成游览。
我正在尝试在 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]
我到底做错了什么?
parseInput
和 buildDistMap
函数以防万一,但它们不是我问题的根源
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\
2 1\n\
4 0\n\
2 0\n\
0 0\n\
4 3\n\
0 3"
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。)
解决这个问题的最佳方法是让 exec
返回的 TS
构造函数的计算依赖于对 visited'
、ordered'
和 [=16 的计算=] 足够了。
M.Map
始终是 spine-strict,因此评估地图意味着评估整个结构。这些值是否也取决于您如何导入它,但事实证明这与此处无关。您要插入的值是空构造函数,因此它已被完全计算。因此,将 visited'
评估为 WHNF 就足够了。
Int
不是嵌套类型,因此将 best
评估为 WHNF 就足够了。
[(Int, Double)]
(外部 parens 是多余的,列表括号对其内容进行分组)有点棘手。列表不是 spine 严格的,pairs 也不是严格的。但是看构造模式,这是一个prepend-only结构。因此,您无需担心尾巴。如果列表被评估进来,只要新的头是,输出就会被评估。不幸的是,这意味着你必须对这对有点小心。它的一半与上面构造的 best
值相同,所以还不错。如果它被评估,它被评估! (尽管这确实表明您不需要在每次迭代时都传递它,您可以只使用 ordered
的前面。)该对的另一半是 Double
,它也是 non-nested,所以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>>
在内存消耗上有什么区别?)
感谢 Carl 非常详细的回答。还要感谢 Daniel 指出缓存大量距离实际上可能会导致我的记忆问题。我假设因为我的代码已经通过了那个函数,所以我有足够的内存来实现它——忘记 Haskell 是懒惰的,只是在我实际使用它时在 exec
函数中构建那个映射。
我现在以更简洁的方式解决了这个问题。我正在使用我仍然需要访问的所有城市索引的 Data.Set
,然后因为城市是按照 X
值的顺序给出的,我知道飞机上最近的城市也是最近的城市指数。知道这一点后,我设置了一个值,以便在每次迭代时从索引的任一侧获取一个切片,并使用该切片检查到我当前城市的距离,这使我可以在每次迭代时计算到下一个城市的距离,而无需缓存大量数据。
-- How many cities in each direction (index) to consider
-- smaller is faster but less accurate
searchWidth = 1000 :: Int
data TS = TS (M.Map Int City) (S.Set Int) [(Double,Int)] Int
run :: String -> String
run input =
let cm = parseInput input
n = length $ M.keys cm
toVisit = S.fromList [1..n]
ts = TS cm toVisit [(0.0,1)] 1
(TS _ _ beforeLast _) = foldl' (\ts i -> trace (concat [show i,"/",show n]) exec ts) ts [2..n]
afterLast = end cm beforeLast
in show $ floor $ sum $ map (\(d,_) -> d) afterLast
exec :: TS -> TS
exec (TS cm toVisit visited curr) =
let (Just (City _ cx cy)) = M.lookup curr cm
index = S.findIndex curr toVisit
toVisit' = S.deleteAt index toVisit
lb = let x = index - searchWidth in if x < 0 then 0 else x
ub = let x = index + searchWidth - lb in if x >= length toVisit' then (length toVisit') else x
candidateIndexes = S.take ub $ S.drop lb toVisit'
candidates = S.map (\i -> let (Just (City _ x y)) = M.lookup i cm in (eDist (x,y) (cx,cy),i)) candidateIndexes
(dist,next) = S.findMin candidates
visited' = (dist,next) : visited
in toVisit' `seq` dist `seq` next `seq` TS cm toVisit' visited' next
end :: M.Map Int City -> [(Double,Int)] -> [(Double,Int)]
end cm visited =
let (_,currI) = head visited
(Just (City _ cx cy)) = M.lookup currI cm
(Just (City _ lx ly)) = M.lookup 1 cm
dist = eDist (cx,cy) (lx,ly)
in (dist,1) : visited
使用 Data.Set
还带来了额外的好处,它会自动对内部的值进行排序,让获得下一个旅行地点变得微不足道。
我意识到这不是世界上最好的 Haskell 代码,我正在做一些顽皮的事情,比如直接从地图查找中匹配 Just
而不是使用 [=16] =] 值。此外,有人向我指出我应该使用记录而不是 data
类型来构造我的 TS