递归 IO 函数中的内存泄漏 - PAP答案

【问题标题】：Memory leak in recursive IO function - PAP递归 IO 函数中的内存泄漏 - PAP
【发布时间】：2017-05-09 11:07:12
【问题描述】：

我编写了一个名为amqp-worker 的库，它提供了一个名为worker 的函数，该函数轮询消息队列（如RabbitMQ）以获取消息，并在找到消息时调用处理程序。然后它回到轮询。

它正在泄漏内存。我已经对其进行了分析，图表显示PAP（部分函数应用程序）是罪魁祸首。 我的代码中的漏洞在哪里？在IO 和forever 循环时如何避免泄漏？

这里有一些相关的功能。 The full source is here.

main :: IO ()
main = do
  -- connect
  conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")

  -- initialize the queues
  Worker.initQueue conn queue
  Worker.initQueue conn results

  -- publish a message
  Worker.publish conn queue (TestMessage "hello world")

  -- create a worker, the program loops here
  Worker.worker def conn queue onError (onMessage conn)

worker

worker :: (FromJSON a, MonadBaseControl IO m, MonadCatch m) => WorkerOptions -> Connection -> Queue key a -> (WorkerException SomeException -> m ()) -> (Message a -> m ()) -> m ()
worker opts conn queue onError action =
  forever $ do
    eres <- consumeNext (pollDelay opts) conn queue
    case eres of
      Error (ParseError reason bd) ->
        onError (MessageParseError bd reason)

      Parsed msg ->
        catch
          (action msg)
          (onError . OtherException (body msg))
    liftBase $ threadDelay (loopDelay opts)

consumeNext

consumeNext :: (FromJSON msg, MonadBaseControl IO m) => Microseconds -> Connection -> Queue key msg -> m (ConsumeResult msg)
consumeNext pd conn queue =
    poll pd $ consume conn queue

poll

poll :: (MonadBaseControl IO m) => Int -> m (Maybe a) -> m a
poll us action = do
    ma <- action
    case ma of
      Just a -> return a
      Nothing -> do
        liftBase $ threadDelay us
        poll us action

【问题讨论】：

你的 ghc 版本是什么，你是如何编译的？
它设置为 lts-7.3，所以这是 GHC 8.0.1。我正在使用 stack install --profile 进行编译。但是我通过正常的堆栈安装得到了内存泄漏。使用堆栈模板中的默认 ghc 选项：-threaded -rtsopts -with-rtsopts=-N
这个例子远非最小——你在你的例子程序中导入你的整个库（Network.AMQP.Worker）。就目前而言，这太宽泛了。
这是我第一次寻找内存泄漏。我想知道是否有一种好方法可以确定它们在这样的程序中的位置
我会尝试摆脱 monad 类型类，只使用 IO。我不确定这是否可能是问题所在，但最好少担心一件事情。

标签： haskell recursion memory-leaks profiling

【解决方案1】：

这里有一个非常简单的例子来说明你的问题：

main :: IO ()
main = worker

{-# NOINLINE worker #-}
worker :: (Monad m) => m ()
worker =
  let loop = poll >> loop
  in loop

poll :: (Monad m) => m a
poll = return () >> poll

如果您删除 NOINLINE，或将 m 专门化为 IO（使用 -O 编译时），泄漏消失了。

我写了一个详细的blog post为什么正是这段代码泄漏了内存。快速总结是，正如里德在他的答案是，代码会创建并记住一连串的部分应用程序 >>s.

我还为此提交了ghc ticket。

【讨论】：

【解决方案2】：

也许一个更容易理解的例子是这个

main :: IO ()
main = let c = count 0
       in c >> c

{-# NOINLINE count #-}
count :: Monad m => Int -> m ()
count 1000000 = return ()
count n = return () >> count (n+1)

为 IO 操作评估 f >> g 会产生某种闭包，该闭包同时引用了 f 和 g（它基本上是 f 和 g 的组合，作为状态令牌上的函数）。 count 0 返回一个 thunk c，它将评估为 return () >> return () >> return () >> ... 形式的大型闭包结构。当我们执行c 时，我们建立了这个结构，因为我们必须第二次执行c，整个结构仍然有效。所以这个程序会泄漏内存（不管优化标志如何）。

当count 专用于IO 并启用优化时，GHC 有多种技巧可用于避免构建此数据结构；但他们都依赖于知道 monad 是 IO。

回到原来的count :: Monad m => Int -> m ()，我们可以通过将最后一行改为

来尽量避免构建这个大结构

count n = return () >>= (\_ -> count (n+1))

现在递归调用隐藏在 lambda 中，所以 c 只是一个小结构 return () >>= (\_ -> BODY)。这确实避免了在没有优化的情况下编译时的空间泄漏。但是，当启用优化时，GHC 会从 lambda 主体中浮出 count (n+1)（因为它不依赖于参数），从而产生

count n = return () >>= (let body = count (n+1) in \_ -> body)

现在c又是一个大结构...

【讨论】：

NOINLINE的使用如何使程序与原来的泄漏程序相媲美？
GHC 不内联或专门化是一般情况（当函数在不同的模块中定义时，不是小等）GHC 知道很多技巧，当你最小化这些技巧时可能会起作用。使用NOINLINE 停止了许多这些技巧，让您进一步最小化。

【解决方案3】：

内存泄漏在poll。使用monad-loops，我将定义更改为以下内容：看起来untilJust 与我的递归执行相同的操作，但修复了泄漏。

谁能评论为什么我之前对poll 的定义会泄漏内存？

{-# LANGUAGE FlexibleContexts #-}

module Network.AMQP.Worker.Poll where

import Control.Concurrent (threadDelay)
import Control.Monad.Trans.Control (MonadBaseControl)
import Control.Monad.Base (liftBase)
import Control.Monad.Loops (untilJust)

poll :: (MonadBaseControl IO m) => Int -> m (Maybe a) -> m a
poll us action = untilJust $ do
    ma <- action
    case ma of
      Just a -> return $ Just a
      Nothing -> do
        liftBase $ threadDelay us
        return Nothing

【讨论】：