Réduire le temps de pause de la collecte des déchets dans un programme Haskell

Question

Nous développons un programme qui reçoit et transmet des "messages", tout en conservant un historique temporaire de ces messages, de sorte qu'il puisse vous indiquer l'historique des messages si vous le demandez. Les messages sont identifiés numériquement, leur taille est typiquement d'environ 1 kilooctet, et nous devons conserver des centaines de milliers de ces messages.

Nous souhaitons optimiser ce programme pour la latence : le temps entre l'envoi et la réception d'un message doit être inférieur à 10 millisecondes.

Le programme est écrit en Haskell et compilé avec GHC. Cependant, nous avons constaté que les pauses du ramassage des ordures sont beaucoup trop longues pour nos exigences en matière de latence : plus de 100 millisecondes dans notre programme réel.

Le programme suivant est une version simplifiée de notre application. Il utilise un Data.Map.Strict pour stocker des messages. Les messages sont ByteString identifiés par un Int . 1 000 000 de messages sont insérés par ordre numérique croissant, et les messages les plus anciens sont continuellement supprimés pour maintenir l'historique à un maximum de 200 000 messages.

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])

Nous avons compilé et exécuté ce programme en utilisant :

$ 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

La mesure importante ici est la "pause maximale" de 0,0515s, soit 51 millisecondes. Nous souhaitons réduire cette valeur d'au moins un ordre de grandeur.

L'expérimentation montre que la durée d'une pause GC est déterminée par le nombre de messages dans l'historique. La relation est à peu près linéaire, ou peut-être super-linéaire. Le tableau suivant montre cette relation. ( Vous pouvez consulter nos tests d'évaluation comparative ici et quelques graphiques ici .)

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

Nous avons expérimenté plusieurs autres variables pour savoir si elles pouvaient réduire cette latence, mais aucune ne fait une grande différence. Parmi ces variables sans importance, on trouve : l'optimisation ( -O , -O2 ) ; options RTS GC ( -G , -H , -A , -c ), le nombre de cœurs ( -N ), des structures de données différentes ( Data.Sequence ), la taille des messages et la quantité de déchets éphémères générés. Le facteur déterminant le plus important est le nombre de messages dans l'historique.

Notre théorie de travail est que les pauses sont linéaires dans le nombre de messages parce que chaque cycle GC doit parcourir toute la mémoire accessible de travail et la copier, qui sont clairement des opérations linéaires.

Questions :

• Cette théorie du temps linéaire est-elle correcte ? La longueur des pauses GC peut-elle être exprimée de cette manière simple, ou la réalité est-elle plus complexe ?
• Si la pause GC est linéaire dans la mémoire de travail, y a-t-il un moyen de réduire les facteurs constants impliqués ?
• Existe-t-il des options pour la GC incrémentielle, ou quelque chose de similaire ? Nous ne voyons que des documents de recherche. Nous sommes tout à fait disposés à échanger le débit contre une latence plus faible.
• Existe-t-il des moyens de "partitionner" la mémoire pour des cycles GC plus petits, autres que la division en plusieurs processus ?

Answer 1

Vous vous en sortez plutôt bien pour avoir un temps de pause de 51 ms avec plus de 200 Mo de données en direct. Le système sur lequel je travaille a un temps de pause maximum plus important avec la moitié de cette quantité de données en direct.

Votre hypothèse est correcte, le temps de pause GC majeur est directement proportionnel à la quantité de données en direct, et malheureusement il n'y a aucun moyen de contourner cela avec GHC tel qu'il est. Nous avons expérimenté la GC incrémentielle dans le passé, mais il s'agissait d'un projet de recherche qui n'a pas atteint le niveau de maturité nécessaire pour l'intégrer dans le GHC publié.

Nous espérons que les régions compactes permettront de résoudre ce problème à l'avenir : https://phabricator.haskell.org/D1264 . C'est une sorte de gestion manuelle de la mémoire où vous compactez une structure dans le tas, et le GC n'a pas à la traverser. Cela fonctionne mieux pour les données à longue durée de vie, mais peut-être sera-t-il assez bon pour être utilisé pour les messages individuels dans votre contexte. Notre objectif est de l'avoir dans GHC 8.2.0.

Si vous êtes dans un environnement distribué et que vous disposez d'un équilibreur de charge, il y a des astuces que vous pouvez utiliser pour éviter de subir le coup de la pause, vous devez vous assurer que l'équilibreur de charge n'envoie pas de requêtes aux machines qui sont sur le point d'effectuer une GC majeure, et bien sûr, vous devez vous assurer que la machine termine la GC même si elle ne reçoit pas de requêtes.

Answer 2

Comme mentionné dans d'autres réponses, le garbage collector dans GHC parcourt les données vivantes, ce qui signifie que plus vous stockez de données à longue durée de vie en mémoire, plus les pauses GC seront longues.

GHC 8.2

Pour surmonter partiellement ce problème, une fonctionnalité appelée régions compactes a été introduit dans GHC-8.2. C'est à la fois une fonctionnalité du système d'exécution de GHC et une bibliothèque qui expose pratique pour travailler avec. La fonctionnalité des régions compactes permet de placer vos données dans un endroit distinct de la mémoire et la GC ne les traversera pas pendant la phase de collecte des déchets. Donc, si vous avez une grande structure que vous voulez garder en mémoire, pensez à utiliser les régions compactes. Cependant, la région compacte elle-même n'a pas mini collecteur d'ordures à l'intérieur, cela fonctionne mieux pour en annexe seulement structures de données, et non quelque chose comme HashMap où vous voulez aussi supprimer des choses. Cependant, vous pouvez surmonter ce problème. Pour plus de détails, consultez l'article de blog suivant :

• Utilisation de régions compactes (par Alexander Vershilov)

GHC 8.10

De plus, depuis GHC-8.10, un nouveau module incrémental à faible latence L'algorithme de garbage collector est implémenté. Il s'agit d'un algorithme GC alternatif qui n'est pas activé par défaut, mais vous pouvez y adhérer si vous le souhaitez. Ainsi, vous pouvez remplacer le GC par défaut par un GC plus récent pour bénéficier automatiquement des fonctionnalités offertes par le GC. régions compactes sans avoir besoin d'emballer et de déballer manuellement. Cependant, le nouveau GC n'est pas une solution miracle et ne résout pas tous les problèmes automatiquement, et il a ses inconvénients. Pour des benchmarks de la nouvelle GC, consultez le dépôt GitHub suivant :

• Benchmarks GC incrémentiels à faible latence

Answer 3

J'ai essayé votre extrait de code avec une approche ringbuffer en utilisant IOVector comme structure de données sous-jacente. Sur mon système (GHC 7.10.3, mêmes options de compilation), cela a entraîné une réduction du temps maximum (la mesure que vous avez mentionnée dans votre OP) de ~22%.

NB. J'ai fait deux hypothèses ici :

1. Une structure de données mutable convient bien au problème (je suppose que le passage des messages implique l'entrée/sortie de toute façon).
2. Vos messages d'identification sont continus

Avec quelques Int et l'arithmétique (comme lorsque les messageId sont remis à 0 ou minBound ), il devrait alors être simple de déterminer si un certain message est toujours dans l'historique et de le récupérer à partir de l'index correspondant dans le tampon circulaire.

Pour votre plaisir de tester :

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])

Answer 4

Je suis d'accord avec les autres : si vous avez des contraintes de temps réel strictes, l'utilisation d'un langage GC n'est pas idéale.

Cependant, vous pourriez envisager d'expérimenter d'autres structures de données disponibles plutôt que de vous limiter à Data.Map.

Je l'ai réécrit en utilisant Data.Sequence et j'ai obtenu quelques améliorations prometteuses :

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

Même si vous optimisez la latence, j'ai remarqué que d'autres paramètres s'améliorent également. Dans le cas de 200000, le temps d'exécution passe de 1,5 s à 0,2 s, et l'utilisation totale de la mémoire passe de 600 Mo à 27 Mo.

Je dois noter que j'ai triché en modifiant le design :

• J'ai retiré le Int de la Msg donc ce n'est pas à deux endroits.
• Au lieu d'utiliser une carte de Int s à ByteString j'ai utilisé un Sequence de ByteString et au lieu d'un Int par message, je pense que cela peut être fait avec un seul Int pour l'ensemble Sequence . En supposant que les messages ne peuvent pas être réorganisés, vous pouvez utiliser un seul décalage pour traduire le message que vous voulez à l'endroit où il se trouve dans la file d'attente.

(J'ai inclus une fonction supplémentaire getMsg pour le démontrer).

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

Answer 5

Vous avez trouvé la limite des langages avec GC : ils ne sont pas adaptés aux systèmes temps réel hardcore.

Vous avez deux options :

1. Augmentez la taille du tas et utilisez un système de cache à 2 niveaux, les messages les plus anciens sont envoyés sur le disque et vous gardez les messages les plus récents en mémoire, vous pouvez le faire en utilisant la pagination du système d'exploitation. Le problème, cependant, avec cette solution est que la pagination peut être coûteuse en fonction des capacités de lecture de l'unité de mémoire secondaire utilisée.

2. Programmez cette solution en utilisant 'C' et interfacez-la avec FFI à haskell. De cette façon, vous pouvez faire votre propre gestion de la mémoire. Ce serait la meilleure option car vous pouvez contrôler vous-même la mémoire dont vous avez besoin.