Reduzindo o tempo de pausa na coleta de lixo em um programa Haskell

Estamos desenvolvendo um programa que recebe e encaminha "mensagens", mantendo um histórico temporário dessas mensagens, para que ele possa lhe informar o histórico, se solicitado. As mensagens são identificadas numericamente, geralmente têm cerca de 1 kilobyte e precisamos manter centenas de milhares dessas mensagens.

Desejamos otimizar este programa para latência: o tempo entre o envio e o recebimento de uma mensagem deve ser inferior a 10 milissegundos.

O programa foi escrito em Haskell e compilado com o GHC. No entanto, descobrimos que as pausas na coleta de lixo são muito longas para nossos requisitos de latência: mais de 100 milissegundos em nosso programa no mundo real.

O programa a seguir é uma versão simplificada do nosso aplicativo. Ele usa a Data.Map.Strictpara armazenar mensagens. As mensagens são ByteStringidentificadas por um Int. 1.000.000 de mensagens são inseridas em ordem numérica crescente e as mensagens mais antigas são continuamente removidas para manter o histórico em um máximo de 200.000 mensagens.

module Main (main) where

import qualified Control.Exception as Exception
import qualified Control.Monad as Monad
import qualified Data.ByteString as ByteString
import qualified Data.Map.Strict as Map

data Msg = Msg !Int !ByteString.ByteString

type Chan = Map.Map Int ByteString.ByteString

message :: Int -> Msg
message n = Msg n (ByteString.replicate 1024 (fromIntegral n))

pushMsg :: Chan -> Msg -> IO Chan
pushMsg chan (Msg msgId msgContent) =
  Exception.evaluate $
    let
      inserted = Map.insert msgId msgContent chan
    in
      if 200000 < Map.size inserted
      then Map.deleteMin inserted
      else inserted

main :: IO ()
main = Monad.foldM_ pushMsg Map.empty (map message [1..1000000])

Compilamos e executamos este programa usando:

$ ghc --version
The Glorious Glasgow Haskell Compilation System, version 7.10.3
$ ghc -O2 -optc-O3 Main.hs
$ ./Main +RTS -s
   3,116,460,096 bytes allocated in the heap
     385,101,600 bytes copied during GC
     235,234,800 bytes maximum residency (14 sample(s))
     124,137,808 bytes maximum slop
             600 MB total memory in use (0 MB lost due to fragmentation)

                                     Tot time (elapsed)  Avg pause  Max pause
  Gen  0      6558 colls,     0 par    0.238s   0.280s     0.0000s    0.0012s
  Gen  1        14 colls,     0 par    0.179s   0.250s     0.0179s    0.0515s

  INIT    time    0.000s  (  0.000s elapsed)
  MUT     time    0.652s  (  0.745s elapsed)
  GC      time    0.417s  (  0.530s elapsed)
  EXIT    time    0.010s  (  0.052s elapsed)
  Total   time    1.079s  (  1.326s elapsed)

  %GC     time      38.6%  (40.0% elapsed)

  Alloc rate    4,780,213,353 bytes per MUT second

  Productivity  61.4% of total user, 49.9% of total elapsed

A métrica importante aqui é a "pausa máxima" de 0,0515s ou 51 milissegundos. Desejamos reduzir isso em pelo menos uma ordem de magnitude.

A experiência mostra que a duração de uma pausa no GC é determinada pelo número de mensagens no histórico. O relacionamento é aproximadamente linear, ou talvez super-linear. A tabela a seguir mostra esse relacionamento. ( Você pode ver nossos testes de benchmarking aqui e alguns gráficos aqui .)

msgs history length  max GC pause (ms)
===================  =================
12500                                3
25000                                6
50000                               13
100000                              30
200000                              56
400000                             104
800000                             199
1600000                            487
3200000                           1957
6400000                           5378

Experimentamos várias outras variáveis ​​para descobrir se elas podem reduzir essa latência, nenhuma das quais faz uma grande diferença. Entre essas variáveis ​​sem importância estão: otimização ( -O, -O2); Opções RTS GC ( -G, -H, -A, -c), número de núcleos ( -N), estruturas de dados diferentes ( Data.Sequence), o tamanho das mensagens, ea quantidade de lixo gerado de curta duração. O fator determinante esmagador é o número de mensagens na história.

Nossa teoria de trabalho é que as pausas são lineares no número de mensagens, porque cada ciclo do GC precisa percorrer toda a memória acessível de trabalho e copiá-la, que são operações claramente lineares.

Questões:

• Essa teoria do tempo linear está correta? A duração das pausas no GC pode ser expressa dessa maneira simples ou a realidade é mais complexa?
• Se a pausa do GC for linear na memória de trabalho, existe alguma maneira de reduzir os fatores constantes envolvidos?
• Existem opções para GC incremental ou algo parecido? Só podemos ver trabalhos de pesquisa. Estamos muito dispostos a trocar a taxa de transferência por menor latência.
• Existem maneiras de "particionar" a memória para ciclos menores de GC, além da divisão em vários processos?

Você está realmente indo muito bem para ter um tempo de pausa de 51ms com mais de 200Mb de dados ao vivo. O sistema no qual trabalho tem um tempo de pausa máximo maior, com metade dessa quantidade de dados ativos.

Sua suposição está correta, o maior tempo de pausa do GC é diretamente proporcional à quantidade de dados ativos e, infelizmente, não há como contornar isso com o GHC como está. Experimentamos GC incremental no passado, mas era um projeto de pesquisa e não alcançava o nível de maturidade necessário para dobrá-lo no GHC liberado.

Uma coisa que esperamos que ajude com isso no futuro são as regiões compactas: https://phabricator.haskell.org/D1264 . É um tipo de gerenciamento manual de memória, no qual você compacta uma estrutura na pilha e o GC não precisa atravessá-la. Funciona melhor para dados de longa duração, mas talvez seja bom o suficiente para usar em mensagens individuais em sua configuração. Nosso objetivo é tê-lo no GHC 8.2.0.

Se você estiver em uma configuração distribuída e tiver um balanceador de carga de algum tipo, existem truques que você pode executar para evitar a ocorrência de pausa, basicamente você garante que o balanceador de carga não envie solicitações para máquinas que estão prestes a faça um GC importante e, claro, verifique se a máquina ainda o conclui, mesmo que não esteja recebendo solicitações.

Eu tentei seu snippet de código com uma abordagem de ringbuffer usando IOVectorcomo estrutura de dados subjacente. No meu sistema (GHC 7.10.3, mesmas opções de compilação), isso resultou em uma redução do tempo máximo (a métrica que você mencionou no seu OP) em ~ 22%.

NB Fiz duas suposições aqui:

1. Uma estrutura de dados mutáveis ​​é adequada para o problema (acho que a passagem de mensagens implica E / S de qualquer maneira)
2. As suas mensagens são contínuas

Com alguns Intparâmetros e aritméticos adicionais (como quando os ID da mensagem são redefinidos para 0 ou minBound), deve ser simples determinar se uma determinada mensagem ainda está no histórico e recuperá-la do índice correspondente no buffer de anel.

Para seu prazer em testar:

import qualified Control.Exception as Exception
import qualified Control.Monad as Monad
import qualified Data.ByteString as ByteString
import qualified Data.Map.Strict as Map

import qualified Data.Vector.Mutable as Vector

data Msg = Msg !Int !ByteString.ByteString

type Chan = Map.Map Int ByteString.ByteString

data Chan2 = Chan2
    { next          :: !Int
    , maxId         :: !Int
    , ringBuffer    :: !(Vector.IOVector ByteString.ByteString)
    }

chanSize :: Int
chanSize = 200000

message :: Int -> Msg
message n = Msg n (ByteString.replicate 1024 (fromIntegral n))


newChan2 :: IO Chan2
newChan2 = Chan2 0 0 <$> Vector.unsafeNew chanSize

pushMsg2 :: Chan2 -> Msg -> IO Chan2
pushMsg2 (Chan2 ix _ store) (Msg msgId msgContent) =
    let ix' = if ix == chanSize then 0 else ix + 1
    in Vector.unsafeWrite store ix' msgContent >> return (Chan2 ix' msgId store)

pushMsg :: Chan -> Msg -> IO Chan
pushMsg chan (Msg msgId msgContent) =
  Exception.evaluate $
    let
      inserted = Map.insert msgId msgContent chan
    in
      if chanSize < Map.size inserted
      then Map.deleteMin inserted
      else inserted

main, main1, main2 :: IO ()

main = main2

main1 = Monad.foldM_ pushMsg Map.empty (map message [1..1000000])

main2 = newChan2 >>= \c -> Monad.foldM_ pushMsg2 c (map message [1..1000000])

Eu tenho que concordar com os outros - se você tiver restrições difíceis em tempo real, usar uma linguagem GC não é o ideal.

No entanto, você pode considerar experimentar outras estruturas de dados disponíveis, e não apenas o Data.Map.

Eu o reescrevi usando Data.Sequence e obtive algumas melhorias promissoras:

msgs history length  max GC pause (ms)
===================  =================
12500                              0.7
25000                              1.4
50000                              2.8
100000                             5.4
200000                            10.9
400000                            21.8
800000                            46
1600000                           87
3200000                          175
6400000                          350

Embora você esteja otimizando a latência, notei que outras métricas também estavam melhorando. No caso 200000, o tempo de execução cai de 1,5s para 0,2s e o uso total de memória cai de 600MB para 27MB.

Devo notar que trapaceie ao ajustar o design:

• Tirei o Intdo Msg, por isso não está em dois lugares.
• Em vez de usar um mapa de Ints para ByteStrings, usei um Sequencede ByteStrings e, em vez de um Intpor mensagem, acho que pode ser feito com um Intpara o todo Sequence. Supondo que as mensagens não possam ser reordenadas, você pode usar um único deslocamento para traduzir qual mensagem deseja para onde fica a fila.

(Incluí uma função adicional getMsgpara demonstrar isso.)

{-# LANGUAGE BangPatterns #-}

import qualified Control.Exception as Exception
import qualified Control.Monad as Monad
import qualified Data.ByteString as ByteString
import Data.Sequence as S

newtype Msg = Msg ByteString.ByteString

data Chan = Chan Int (Seq ByteString.ByteString)

message :: Int -> Msg
message n = Msg (ByteString.replicate 1024 (fromIntegral n))

maxSize :: Int
maxSize = 200000

pushMsg :: Chan -> Msg -> IO Chan
pushMsg (Chan !offset sq) (Msg msgContent) =
    Exception.evaluate $
        let newSize = 1 + S.length sq
            newSq = sq |> msgContent
        in
        if newSize <= maxSize
            then Chan offset newSq
            else
                case S.viewl newSq of
                    (_ :< newSq') -> Chan (offset+1) newSq'
                    S.EmptyL -> error "Can't happen"

getMsg :: Chan -> Int -> Maybe Msg
getMsg (Chan offset sq) i_ = getMsg' (i_ - offset)
    where
    getMsg' i
        | i < 0            = Nothing
        | i >= S.length sq = Nothing
        | otherwise        = Just (Msg (S.index sq i))

main :: IO ()
main = Monad.foldM_ pushMsg (Chan 0 S.empty) (map message [1..5 * maxSize])

Como mencionado em outras respostas, o coletor de lixo no GHC percorre os dados ao vivo, o que significa que quanto mais dados você armazenar na memória, maiores serão as pausas no GC.

GHC 8.2

Para superar esse problema parcialmente, um recurso chamado regiões compactas foi introduzido no GHC-8.2. É um recurso do sistema de tempo de execução GHC e uma biblioteca que expõe uma interface conveniente para trabalhar. O recurso de regiões compactas permite colocar seus dados em um local separado na memória e o GC não os percorrerá durante a fase de coleta de lixo. Portanto, se você tem uma estrutura grande que deseja manter na memória, considere usar regiões compactas. No entanto, a região compacta em si não possui um mini coletor de lixo , ela funciona melhor para estruturas de dados somente anexadas, não como algo HashMaponde você também deseja excluir coisas. Embora você possa superar esse problema. Para detalhes, consulte a seguinte postagem no blog:

• Usando regiões compactas (por Alexander Vershilov)

GHC 8.10

Além disso, desde o GHC-8.10, um novo algoritmo de coletor de lixo incremental de baixa latência é implementado. É um algoritmo de GC alternativo que não está ativado por padrão, mas você pode ativá-lo, se desejar. Assim, você pode mudar o GC padrão para um mais novo para obter automaticamente os recursos fornecidos por regiões compactas sem a necessidade de empacotamento e desembrulhamento manual. No entanto, o novo GC não é uma bala de prata e não resolve todos os problemas automaticamente, e tem suas vantagens e desvantagens. Para referências do novo GC, consulte o seguinte repositório GitHub:

• Benchmarks incrementais de GC de baixa latência

Bem, você encontrou a limitação de idiomas com o GC: eles não são adequados para sistemas hardcore em tempo real.

Você tem 2 opções:

1º Aumente o tamanho da pilha e use um sistema de armazenamento em cache de 2 níveis, as mensagens mais antigas são enviadas para o disco e você mantém as mensagens mais recentes na memória; você pode fazer isso usando a paginação do SO. O problema, embora com esta solução, é que a paginação pode ser cara, dependendo dos recursos de leitura da unidade de memória secundária usada.

2º Programe essa solução usando 'C' e faça a interface com a FFI para haskell. Dessa forma, você pode fazer seu próprio gerenciamento de memória. Essa seria a melhor opção, pois você pode controlar a memória de que precisa.