Reduzieren der Pausenzeit für die Speicherbereinigung in einem Haskell-Programm

Wir entwickeln ein Programm, das "Nachrichten" empfängt und weiterleitet und dabei einen temporären Verlauf dieser Nachrichten führt, damit es Ihnen auf Anfrage den Nachrichtenverlauf mitteilen kann. Nachrichten werden numerisch identifiziert, haben normalerweise eine Größe von etwa 1 Kilobyte und wir müssen Hunderttausende dieser Nachrichten aufbewahren.

Wir möchten dieses Programm auf Latenz optimieren: Die Zeit zwischen dem Senden und Empfangen einer Nachricht muss unter 10 Millisekunden liegen.

Das Programm ist in Haskell geschrieben und mit GHC kompiliert. Wir haben jedoch festgestellt, dass die Speicherbereinigungspausen für unsere Latenzanforderungen viel zu lang sind: Über 100 Millisekunden in unserem realen Programm.

Das folgende Programm ist eine vereinfachte Version unserer Anwendung. Es verwendet a Data.Map.Strictzum Speichern von Nachrichten. Nachrichten werden ByteStringdurch ein gekennzeichnet Int. 1.000.000 Nachrichten werden in aufsteigender numerischer Reihenfolge eingefügt, und die ältesten Nachrichten werden kontinuierlich entfernt, um den Verlauf auf maximal 200.000 Nachrichten zu beschränken.

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

Wir haben dieses Programm kompiliert und ausgeführt mit:

$ 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

Die wichtige Metrik hier ist die "maximale Pause" von 0,0515 s oder 51 Millisekunden. Wir wollen dies um mindestens eine Größenordnung reduzieren.

Experimente zeigen, dass die Länge einer GC-Pause durch die Anzahl der Nachrichten im Verlauf bestimmt wird. Die Beziehung ist ungefähr linear oder vielleicht superlinear. Die folgende Tabelle zeigt diese Beziehung. ( Sie können unsere Benchmarking-Tests hier und einige Diagramme hier sehen .)

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

Wir haben mit mehreren anderen Variablen experimentiert, um herauszufinden, ob sie diese Latenz reduzieren können, von denen keine einen großen Unterschied macht. Zu diesen unwichtigen Variablen gehören: Optimierung ( -O, -O2); RTS GC - Optionen ( -G, -H, -A, -c), Anzahl der Kerne ( -N), verschiedene Datenstrukturen ( Data.Sequence), die Größe der Nachrichten, und die Menge an erzeugtem kurzlebig Müll. Der überwältigende bestimmende Faktor ist die Anzahl der Nachrichten im Verlauf.

Unsere Arbeitstheorie besagt, dass die Pausen in der Anzahl der Nachrichten linear sind, da jeder GC-Zyklus den gesamten arbeitssicheren Speicher durchlaufen und kopieren muss, was eindeutig lineare Operationen sind.

Fragen:

• Ist diese lineare Zeittheorie richtig? Kann die Länge der GC-Pausen auf diese einfache Weise ausgedrückt werden oder ist die Realität komplexer?
• Wenn die GC-Pause im Arbeitsspeicher linear ist, gibt es eine Möglichkeit, die konstanten Faktoren zu reduzieren?
• Gibt es Optionen für inkrementelle GC oder ähnliches? Wir können nur Forschungsarbeiten sehen. Wir sind sehr bereit, den Durchsatz gegen eine geringere Latenz zu tauschen.
• Gibt es andere Möglichkeiten, den Speicher für kleinere GC-Zyklen zu "partitionieren", als ihn in mehrere Prozesse aufzuteilen?

Sie machen es tatsächlich ziemlich gut, eine Pausenzeit von 51 ms mit über 200 MB Live-Daten zu haben. Das System, an dem ich arbeite, hat eine größere maximale Pausenzeit mit der Hälfte dieser Menge an Live-Daten.

Ihre Annahme ist richtig, die Haupt-GC-Pausenzeit ist direkt proportional zur Menge der Live-Daten, und leider führt bei GHC kein Weg daran vorbei. Wir haben in der Vergangenheit mit inkrementeller GC experimentiert, aber es war ein Forschungsprojekt und erreichte nicht den Reifegrad, der erforderlich war, um es in die freigegebene GHC zu falten.

Eine Sache, von der wir hoffen, dass sie in Zukunft dabei helfen wird, sind kompakte Regionen: https://phabricator.haskell.org/D1264 . Es ist eine Art manuelle Speicherverwaltung, bei der Sie eine Struktur im Heap komprimieren und der GC sie nicht durchlaufen muss. Es funktioniert am besten für langlebige Daten, ist aber möglicherweise gut genug, um es für einzelne Nachrichten in Ihrer Umgebung zu verwenden. Wir wollen es in GHC 8.2.0 haben.

Wenn Sie sich in einer verteilten Umgebung befinden und über einen Load-Balancer verfügen, können Sie Tricks anwenden, um den Pausentreffer zu vermeiden. Stellen Sie im Grunde sicher, dass der Load-Balancer keine Anforderungen an Computer sendet, die in Kürze verfügbar sind Führen Sie einen größeren GC durch, und stellen Sie natürlich sicher, dass der Computer den GC weiterhin abschließt, obwohl keine Anforderungen eingehen.

Ich habe Ihr Code-Snippet mit einem Ringbuffer-Ansatz IOVectorals zugrunde liegende Datenstruktur ausprobiert . Auf meinem System (GHC 7.10.3, gleiche Kompilierungsoptionen) führte dies zu einer Reduzierung der maximalen Zeit (der Metrik, die Sie in Ihrem OP erwähnt haben) um ~ 22%.

NB. Ich habe hier zwei Annahmen getroffen:

1. Eine veränderbare Datenstruktur ist für das Problem in Ordnung (ich denke, das Weiterleiten von Nachrichten impliziert ohnehin E / A).
2. Ihre Nachrichten-IDs sind fortlaufend

Mit einigen zusätzlichen IntParametern und Arithmetiken (z. B. wenn Nachrichten-IDs auf 0 oder zurückgesetzt werden minBound) sollte es dann einfach sein, festzustellen, ob sich eine bestimmte Nachricht noch im Verlauf befindet, und sie aus dem entsprechenden Index im Ringpuffer abzurufen.

Für Ihr Testvergnügen:

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

Ich muss den anderen zustimmen - wenn Sie harte Echtzeitbeschränkungen haben, ist die Verwendung einer GC-Sprache nicht ideal.

Sie können jedoch auch mit anderen verfügbaren Datenstrukturen experimentieren und nicht nur mit Data.Map.

Ich habe es mit Data.Sequence umgeschrieben und einige vielversprechende Verbesserungen erhalten:

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

Obwohl Sie die Latenz optimieren, habe ich festgestellt, dass sich auch andere Metriken verbessert haben. Im Fall 200000 sinkt die Ausführungszeit von 1,5 s auf 0,2 s und die Gesamtspeicherauslastung von 600 MB auf 27 MB.

Ich sollte beachten, dass ich betrogen habe, indem ich das Design optimiert habe:

• Ich habe das Intaus dem entfernt Msg, es ist also nicht an zwei Stellen.
• Anstatt eine Karte von Ints nach ByteStrings zu verwenden, habe ich eine Sequencevon ByteStrings verwendet, und anstelle einer Karte Intpro Nachricht denke ich, dass dies mit einer Intfür das Ganze möglich ist Sequence. Angenommen, Nachrichten können nicht neu angeordnet werden, können Sie einen einzelnen Versatz verwenden, um die gewünschte Nachricht in die Warteschlange zu übersetzen.

(Ich habe eine zusätzliche Funktion getMsghinzugefügt, um dies zu demonstrieren.)

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

Wie in anderen Antworten erwähnt, durchläuft der Garbage Collector in GHC Live-Daten. Je mehr langlebige Daten Sie im Speicher speichern, desto länger sind die GC-Pausen.

GHC 8.2

Um dieses Problem teilweise zu lösen, wurde in GHC-8.2 eine Funktion eingeführt, die als kompakte Regionen bezeichnet wird . Es ist sowohl eine Funktion des GHC-Laufzeitsystems als auch eine Bibliothek, die eine praktische Schnittstelle für die Arbeit bietet. Die Funktion für kompakte Regionen ermöglicht es, Ihre Daten an einem separaten Ort im Speicher abzulegen, und GC durchläuft sie während der Speicherbereinigungsphase nicht. Wenn Sie also eine große Struktur haben, die Sie im Speicher behalten möchten, sollten Sie kompakte Regionen verwenden. In der kompakten Region selbst befindet sich jedoch kein Mini-Garbage-Collector . Sie eignet sich besser für reine Datenstrukturen zum Anhängen , nicht für Bereiche , in HashMapdenen Sie auch Inhalte löschen möchten. Sie können dieses Problem jedoch überwinden. Einzelheiten finden Sie im folgenden Blog-Beitrag:

• Verwendung kompakter Regionen (von Alexander Vershilov)

GHC 8.10

Darüber hinaus ist seit GHC-8.10 ein neuer inkrementeller Garbage Collector-Algorithmus mit geringer Latenz implementiert. Es ist ein alternativer GC-Algorithmus, der nicht standardmäßig aktiviert ist, aber Sie können ihn aktivieren, wenn Sie möchten. Sie können also den Standard-GC auf einen neueren umstellen, um automatisch Funktionen zu erhalten, die von kompakten Regionen bereitgestellt werden , ohne dass manuelles Ein- und Auspacken erforderlich ist. Der neue GC ist jedoch kein Wundermittel und löst nicht alle Probleme automatisch und hat seine Kompromisse. Benchmarks für den neuen GC finden Sie im folgenden GitHub-Repository:

• Inkrementelle GC-Benchmarks mit geringer Latenz

Nun, Sie haben die Einschränkung von Sprachen mit GC gefunden: Sie sind nicht für Hardcore-Echtzeitsysteme geeignet.

Sie haben 2 Möglichkeiten:

1. Erhöhen Sie die Heap-Größe und verwenden Sie ein 2-Level-Caching-System. Die ältesten Nachrichten werden an die Festplatte gesendet, und Sie behalten die neuesten Nachrichten im Speicher. Sie können dies mithilfe von OS-Paging tun. Das Problem bei dieser Lösung ist jedoch, dass das Paging abhängig von den Lesefähigkeiten der verwendeten sekundären Speichereinheit teuer sein kann.

2. Programmieren Sie diese Lösung mit 'C' und verbinden Sie sie mit FFI mit haskell. Auf diese Weise können Sie Ihre eigene Speicherverwaltung durchführen. Dies ist die beste Option, da Sie den benötigten Speicher selbst steuern können.