haskell - recursivo - Pérdida de memoria en la función de IO recursiva-PAP

recursividad estructura de datos

Question

He escrito una biblioteca llamada amqp-worker que proporciona una función llamada worker que sondea una cola de mensajes (como RabbitMQ ) en busca de mensajes, llamando a un manejador cuando se encuentra un mensaje. Luego vuelve al sondeo.

Se está perdiendo la memoria. Lo he perfilado y el gráfico dice que PAP (aplicación de función parcial) es el culpable. ¿Dónde está la fuga en mi código? ¿Cómo puedo evitar las fugas cuando se realiza un bucle en IO forever ?

Aquí hay algunas funciones relevantes. La fuente completa está aquí .

Programa de ejemplo . Esto gotea

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

Answer 1

Aquí hay un ejemplo muy simple que demuestra su problema:

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
If you remove the `NOINLINE`, or specialize `m` to
`IO` (while compiling with `-O`), the leak goes away.

Escribí una publicación detallada del blog sobre por qué exactamente este código pierde memoria. El resumen rápido es, como Reid señala en su respuesta, que el código crea y recuerda una cadena de aplicaciones parciales de >> s.

También presenté un boleto de ghc sobre esto.

Answer 2

La pérdida de memoria estaba en la poll . Usando monad-loops , cambié la definición a lo siguiente: Parece que untilJust que untilJust hace lo mismo que mi recursión, pero corrige la fuga.

¿Alguien puede comentar por qué mi definición anterior de poll estaba perdiendo memoria?

{-# 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

Answer 3

Tal vez un ejemplo más fácil de entender es este

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)

La evaluación de f >> g para las acciones de IO produce algún tipo de cierre que tiene referencias tanto a f como a g (es básicamente la composición de f y g como funciones en tokens de estado). count 0 devuelve un thunk c que evaluará una gran estructura de cierres del formulario return () >> return () >> return () >> ... Cuando ejecutamos c , construimos esta estructura, y como tenemos que ejecutar c una segunda vez, toda la estructura aún está activa. Así que este programa pierde memoria (independientemente de los indicadores de optimización).

Cuando el count está especializado en IO y las optimizaciones están habilitadas, GHC tiene una variedad de trucos disponibles para evitar la construcción de esta estructura de datos; pero todos confían en saber que la mónada es IO .

Volviendo al count :: Monad m => Int -> m () original count :: Monad m => Int -> m () , podemos intentar evitar la construcción de esta gran estructura cambiando la última línea a

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

Ahora, la llamada recursiva está oculta dentro de un lambda, por lo que c es solo un return () >>= (/_ -> BODY) estructura pequeña return () >>= (/_ -> BODY) . Esto realmente evita la pérdida de espacio al compilar sin optimizaciones. Sin embargo, cuando las optimizaciones están habilitadas, el GHC flota count (n+1) en el cuerpo de la lambda (ya que no depende del argumento) produciendo

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

y ahora c es una gran estructura de nuevo ...