haskell - 为什么在添加 `bracketOnError` 时这段代码内存泄漏？

Question

首先，我很抱歉没有一个最小的例子(我可以尝试构建一个，但现在我有一个“之前和之后”的例子):
首先是具有内存泄漏的“之后”:

protoReceiver :: RIO FdsEnv ()
protoReceiver = do
  logItS Info ["Entering FarmPCMessage protoReceiver"]
  tMap <- liftIO $ newThreadMap
  fdsEnv <- ask
  let lgr = fdsLogger fdsEnv
  loopBody <- pure $ bracketOnError
    (runResourceT $ protoServe fdsEnv tMap readFarmPCMessage)
    (\(_,w) -> do
      logLogItS Debug lgr ["Entering cleanup for protoReceiver"]
      )
    (\(server,_) -> do
      logLogItS Debug lgr ["Entering FarmPCMessage protoReceiver bracket"]
      server
        .| mapMC (liftIO . traverse_ (persistFarmEntry fdsEnv))
        .| mapMC ((logLogIt Info lgr) . pure)
        .| sinkUnits & runConduitRes
      )
  liftIO loopBody

这是“之前”的代码，它没有内存泄漏:

protoReceiver :: RIO FdsEnv ()
protoReceiver = do
  logItS Info ["Entering FarmPCMessage protoReceiver"]
  tMap <- liftIO $ newThreadMap
  fdsEnv <- ask
  let lgr = fdsLogger fdsEnv

  (dmgrProtoServe, tcpWorker) <- liftIO $ runResourceT
    $ protoServe fdsEnv tMap readFarmPCMessage
  liftIO $ runResourceT $ dmgrProtoServe
    .| mapMC (liftIO . traverse_ (persistFarmEntry fdsEnv))
    .| mapMC ((logLogIt Info lgr) . pure)
    .| sinkUnits & runConduit

我对泄漏进行了一些分析，但我不确定它是否特别有用(对更好的分析图表的任何建议表示赞赏):

Answer 1

问题是经典泄漏场景的一种变体，其中我们在消耗惰性列表时保留对它的头部的引用:

import Data.Foldable (traverse_)

main :: IO ()
main = do
    let xs = [1..]
    traverse_ print xs 
    traverse_ print xs -- commenting this statement solves the leak 

在这里，导管 Source正在充当各种“懒惰列表”。即使在我们使用它时，我们也需要保留对原始源值( server )的引用，因为它必须在发生错误时传递给异常处理程序。然而异常处理程序似乎并没有使用它。
解决方案是在我们传递给 bracketOnError 的主要计算后立即删除该引用。掌握值(value)。我们可以使用 MVar为了这个目的。不是因为它的同步能力，而是因为它是一个可以“留空”的可变引用。
分配操作，而不是返回 (Source m r, a)值，可以返回 (MVar (Source m r), a)取而代之的值(value)。然后，主要计算将执行 takeMVar掌握导管源。一旦我们开始使用源，原始值将被垃圾收集，因为不会再有对它的引用。
以下是 OP 在遵循这些建议后使用的工作代码:

protoReceiver :: RIO FdsEnv ()
protoReceiver = retryForever $ do
  logItS Info ["Entering FarmPCMessage protoReceiver"]
  tMap <- liftIO $ newThreadMap
  fdsEnv <- ask
  let lgr = fdsLogger fdsEnv
  loopBody <- pure $ bracket
    (runResourceT $ do
      swTup <- protoServe fdsEnv tMap readFarmPCMessage
      serverMVar <- newMVar $ fst swTup
      pure (serverMVar, snd $! swTup)
      )
    (\(_, worker) -> do
      logLogItS Debug lgr ["Entering cleanup for protoReceiver"]
      killChildThreads tMap
      cancel worker
      )
    (\(serverMVar, _) -> do
      logLogItS Debug lgr ["Entering FarmPCMessage protoReceiver bracket"]
      server <- takeMVar serverMVar
      logLogItS Debug lgr ["FarmPCMessage protoReceiver bracket: got server"]
      server
        .| mapMC (liftIO . traverse_ (persistFarmEntry fdsEnv))
        .| mapMC ((logLogIt Info lgr) . pure)
        .| sinkUnits & runConduitRes
      )
  liftIO $ retryForever $ loopBody
  where
    killChildThreads = liftIO . killThreadHierarchy