Riesige Speicherverbrauch für einfache Multithread-Haskell

Question

Ich habe ein relativ einfaches "Kopieren" -Programm, das nur alle Zeilen einer Datei in eine andere kopiert. Ich spiele mit TMQueue mit Haskells Gleichzeitigkeit Unterstützung um und STM so dachte ich, dass ich es so versuchen würde:

{-# LANGUAGE BangPatterns #-} 

module Main where 

import Control.Applicative 
import Control.Concurrent.Async    -- from async 
import Control.Concurrent.Chan 
import Control.Concurrent.STM (atomically) 
import Control.Concurrent.STM.TMQueue  -- from stm-chans 
import Control.Monad (replicateM, forM_, forever, unless) 
import qualified Data.ByteString.Char8 as B 
import Data.Function (fix) 
import Data.Maybe (catMaybes, maybe) 
import System.IO (withFile, IOMode(..), hPutStrLn, hGetLine) 
import System.IO.Error (catchIOError) 

input = "data.dat" 
output = "out.dat" 
batch = 100 :: Int 

consumer :: TMQueue B.ByteString -> IO() 
consumer q = withFile output WriteMode $ \fh -> fix $ \loop -> do 
    !items <- catMaybes <$> replicateM batch readitem 
    forM_ items $ B.hPutStrLn fh 
    unless (length items < batch) loop 
    where 
    readitem = do 
     !item <- atomically $ readTMQueue q 
     return item 

producer :: TMQueue B.ByteString -> IO() 
producer q = withFile input ReadMode $ \fh -> 
    (forever (B.hGetLine fh >>= atomically . writeTMQueue q)) 
    `catchIOError` const (atomically (closeTMQueue q) >> putStrLn "Done") 

main :: IO() 
main = do 
    q <- atomically newTMQueue 
    thread <- async $ consumer q 
    producer q 
    wait thread 

Ich mag diese

ghc -e 'writeFile "data.dat" (unlines (map show [1..5000000]))' 

eine kleine Test-Eingabedatei machen und baue es mag diese

ghc --make QueueTest.hs -O2 -prof -auto-all -caf-all -threaded -rtsopts -o q 

Wenn ich es wie so ./q +RTS -s -prof -hc -L60 -N2 laufen, heißt es, dass „2117 MB Gesamtspeicher im Einsatz“! Aber die Eingabedatei ist nur 38 MB!

Ich bin neu im Profiling, aber ich habe Graph nach Graphen erstellt und kann meinen Fehler nicht lokalisieren.

Answer 1

Wie das OP darauf hinweist, kann ich jetzt auch eine richtige Antwort schreiben. Beginnen wir mit dem Speicherverbrauch.

Zwei nützliche Referenzen sind Memory footprint of Haskell data types und http://blog.johantibell.com/2011/06/memory-footprints-of-some-common-data.html. Wir müssen uns auch die Definitionen einiger unserer Strukturen ansehen.

-- from http://hackage.haskell.org/package/stm-chans-3.0.0.2/docs/src/Control-Concurrent-STM-TMQueue.html 

data TMQueue a = TMQueue 
    {-# UNPACK #-} !(TVar Bool) 
    {-# UNPACK #-} !(TQueue a) 
    deriving Typeable 


-- from http://hackage.haskell.org/package/stm-2.4.3/docs/src/Control-Concurrent-STM-TQueue.html 

-- | 'TQueue' is an abstract type representing an unbounded FIFO channel. 
data TQueue a = TQueue {-# UNPACK #-} !(TVar [a]) 
         {-# UNPACK #-} !(TVar [a]) 

TQueue Die Implementierung verwendet eine Standard-Funktions Warteschlange mit einem Lese Ende und Ende schreiben.

Lassen Sie uns eine obere Grenze für die Speichernutzung setzen und davon ausgehen, dass wir die gesamte Datei in die TMQueue lesen, bevor der Verbraucher etwas macht. In diesem Fall enthält das Schreibende unserer TQueue eine Liste mit einem Element pro Eingabezeile (gespeichert als eine Bytefolge). Jede Liste Knoten wird wie folgt aussehen

(:) bytestring tail 

die 3 Wörtern nimmt (1 pro Feld + 1 für den Konstruktor). Jeder bytestring ist 9 Wörter, addieren Sie also die zwei zusammen und es gibt 12 Wörter Overhead pro Linie, die tatsächlichen Daten nicht eingeschlossen. Ihre Testdaten sind 5 Millionen Zeilen, also 60 Millionen Worte Overhead für die gesamte Datei (plus einige Konstanten), was bei einem 64-Bit-System etwa 460MB ist (vorausgesetzt, ich habe meine Berechnungen richtig gemacht, immer fragwürdig). Fügen Sie 40 MB für die tatsächlichen Daten hinzu, und wir erhalten Werte, die denen auf meinem System sehr ähnlich sind.

Also, warum ist unser Speicherverbrauch nahe dieser oberen Grenze? Ich habe eine Theorie (Untersuchung als Übung übrig gelassen!). Erstens läuft der Produzent wahrscheinlich etwas schneller als der Konsument, weil das Lesen normalerweise schneller ist als das Schreiben (ich benutze rotierende Festplatten, vielleicht wäre eine SSD anders). Hier ist die Definition von readTQueue:

-- |Read the next value from the 'TQueue'. 
readTQueue :: TQueue a -> STM a 
readTQueue (TQueue read write) = do 
    xs <- readTVar read 
    case xs of 
    (x:xs') -> do writeTVar read xs' 
        return x 
    [] -> do ys <- readTVar write 
      case ys of 
       [] -> retry 
       _ -> case reverse ys of 
         [] -> error "readTQueue" 
         (z:zs) -> do writeTVar write [] 
            writeTVar read zs 
            return z 

Zuerst versuchen wir, aus dem gelesenen Ende zu lesen, und wenn die leer ist versuchen wir von dem Schreibende zu lesen, nachdem diese Liste rückgängig zu machen.

Was ich denke passiert ist, dass: wenn der Verbraucher von der Schreibende lesen muss, muss es die Eingabeliste innerhalb der STM-Transaktion durchlaufen. Das braucht etwas Zeit, damit es mit dem Produzenten konkurrieren kann. Wenn der Produzent weiter kommt, wird diese Liste länger, was dazu führt, dass das Lesen noch mehr Zeit benötigt, während der der Produzent mehr Werte schreiben kann, was dazu führt, dass der Lesevorgang fehlschlägt. Dieser Prozess wird wiederholt, bis der Produzent fertig ist, und erst dann erhält der Verbraucher die Möglichkeit, den Großteil der Daten zu verarbeiten.Dies ruiniert nicht nur die Parallelität, sondern fügt auch mehr CPU-Overhead hinzu, da die Consumer-Transaktion ständig wiederholt und fehlschlägt.

Also, was ist mit Unagi? Es gibt ein paar wichtige Unterschiede. Erstens verwendet Unagi-Chan Arrays intern statt Listen. Dies reduziert den Aufwand ein wenig. Der meiste Overhead stammt von den ByteString-Zeigern, also nicht viel, aber ein wenig. Zweitens hält Unagi Brocken von Arrays. Selbst wenn wir pessimistisch davon ausgehen, dass der Produzent immer Konflikte gewinnt, wird er, nachdem das Array gefüllt ist, von der Produzentenseite des Kanals verschoben. Jetzt schreibt der Producer in ein neues Array und der Consumer liest aus dem alten Array. Diese Situation ist nahezu ideal; es gibt keine Konkurrenz zu freigegebenen Ressourcen, der Konsument hat eine gute Fundstelle, und da der Konsument an einem anderen Speicherstück arbeitet, gibt es keine Probleme mit der Cache-Kohärenz. Im Gegensatz zu meiner theoretischen Beschreibung der TMQueue, erhalten Sie jetzt gleichzeitige Operationen, die es dem Hersteller erlauben, etwas von der Speicherbelegung zu löschen, so dass er niemals die obere Grenze erreicht.

Nebenbei denke ich, dass die Verbraucher Batching nicht vorteilhaft ist. Die Handles werden bereits vom IO-Subsystem gepuffert, daher glaube ich nicht, dass dies etwas bringt. Bei mir hat sich die Performance ein wenig verbessert, als ich den Consumer auf Line-by-Line umgestellt habe.

Nun, was können Sie für dieses Problem tun? Ausgehend von meiner Arbeitshypothese, dass TMQueue unter Konkurrenzproblemen und Ihren spezifizierten Anforderungen leidet, müssen Sie nur einen anderen Queue-Typ verwenden. Offensichtlich funktioniert Unagi ziemlich gut. Ich versuchte auch TMChan, es war ungefähr 25% langsamer als unagi, aber verwendete 45% weniger Speicher, also könnte eine gute Wahl auch sein. (Dies ist nicht allzu überraschend, TMChan eine andere Struktur von TMQueue hat so wird es unterschiedliche Leistungsmerkmale hat)

Sie auch Ihren Algorithmus könnten versuchen zu ändern, so dass die Hersteller mehrzeiligen Brocken senden. Dies würde den Speicheraufwand von allen ByteStrings verringern.

Also, wann ist es in Ordnung, TMQueue zu verwenden? Wenn der Hersteller und der Verbraucher ungefähr gleich schnell sind oder der Verbraucher schneller ist, sollte es in Ordnung sein. Auch wenn die Verarbeitungszeiten nicht einheitlich sind oder der Hersteller in Bursts läuft, erhalten Sie wahrscheinlich eine gute amortisierte Leistung. Dies ist so ziemlich eine Worst-Case-Situation, und vielleicht sollte es als ein Fehler gegen stm gemeldet werden? Ich denke, wenn die Lesefunktion in

geändert würde, würde dieses Problem vermeiden. Nun sollten die Bindungen z und zs beide lazily ausgewertet werden, so dass die Liste Traversal außerhalb dieser Transaktion passieren würde, so dass der Lesevorgang erfolgreich manchmal unter Konkurrenzbedingungen. Vorausgesetzt, dass ich das Problem natürlich von Anfang an richtig finde (und dass diese Definition faul genug ist). Es könnte jedoch auch andere unerwartete Nachteile geben.