forConcurrently Leistung als sequenzielle

Question

Ich habe eine Funktion geschrieben, dass up-Proben eine Datei von 48 kHz bis 192 kHz mittels eines Filters:

upsample :: Coefficients -> FilePath -> IO() 

Es nimmt die Filterkoeffizienten, die den Pfad der Datei (das muss upsamples sein) und schreibt das Ergebnis in eine neue Datei.

Ich muss up-Probe viele Dateien, so dass ich eine Funktion geschrieben haben ein komplettes Verzeichnis parallel up-Probe, mit forConcurrently_ von Control.Concurrent.Async:

upsampleDirectory :: Directory -> FilePath -> IO() 
upsampleDirectory dir coefPath = do 
    files <- getAllFilesFromDirectory dir 
    coefs <- loadCoefficients coefPath 
    forConcurrently_ files $ upsample coefs 

ich mit der -threaded Option bin Kompilieren und läuft mit +RTS -N2. Was ich sehe, ist, dass das Upsampling von 2 Dateien sequentiell schneller ist als das parallele Upsampling beider Dateien.

Upsampling file1.wav dauert 18.863s. Upsampling file2.wav dauert 18.707s. Das Upsampling eines Verzeichnisses mit file1.wav und file2.wav dauert 66.250s.

Was mache ich falsch?

Ich habe versucht, diesen Beitrag prägnant zu halten, also fragen Sie mich, wenn Sie weitere Einzelheiten zu einigen der Funktionen benötigen.

Answer 1

Hier sind ein paar Möglichkeiten. Machen Sie sich zunächst 100% sicher, dass Sie Ihr Programm tatsächlich mit +RTS -N2 -RTS ausführen. Ich kann Ihnen nicht sagen, wie oft ich habe ein paralleles Programm Benchmarking und geschrieben:

stack exec myprogram +RTS -N2 -RTS 

anstelle von:

stack exec myprogram -- +RTS -N2 -RTS 

und bekommen mich hoffnungslos verwirrt. (Die erste Version führt die Stack ausführbare Datei auf zwei Prozessoren, aber die Ziel-ausführbare Datei auf einem!) Fügen Sie möglicherweise eine print $ getNumCapabilities am Anfang Ihres main-Programms, um sicher zu sein.

Nachdem Sie bestätigt haben, dass Sie auf zwei Prozessoren ausgeführt werden, ist das nächste wahrscheinlichste Problem, dass Ihre Implementierung nicht im konstanten Bereich ausgeführt wird und den Heap in die Luft jagt. Hier ist ein einfaches Testprogramm, mit dem ich versucht habe, Ihr Problem zu reproduzieren. (Fühlen Sie sich frei my awesome Upsampling zu verwenden, um sich filtern!)

module Main where 

import Control.Concurrent.Async 
import System.Environment 
import qualified Data.ByteString as B 

upsample :: FilePath -> IO() 
upsample fp = do c <- B.readFile fp 
       let c' = B.pack $ concatMap (replicate 4) $ B.unpack c 
       B.writeFile (fp ++ ".out") c' 

upsampleFiles :: [FilePath] -> IO() 
upsampleFiles files = do 
    forConcurrently_ files $ upsample 

main :: IO() 
main = upsampleFiles =<< getArgs -- sample all file on command line 

Als ich das lief auf einem einzigen 70meg Testdatei, es in 14 Sekunden lief. Als ich es auf zwei Kopien parallel lief, lief es für mehr als eine Minute, bevor es begann, wie verrückt zu tauschen, und ich musste es töten. Nach dem Umschalten auf:

import qualified Data.ByteString.Lazy as B 

lief es in 3,7 Sekunden auf einer einzige Datei, 7,8 Sekunden auf zwei Kopien auf einem einzelnen Prozessor und 4,0 Sekunden auf zwei Kopien auf zwei Prozessoren mit +RTS -N2.

Stellen Sie sicher, dass Sie mit Optimierungen arbeiten, Ihr Programm profilieren und sicherstellen, dass es in einem konstanten (oder zumindest angemessenen) Heap-Bereich läuft. Das obige Programm wird in einem konstanten Heap von 100 KB ausgeführt.Eine ähnliche Version, die eine strenge ByteString zum Lesen und eine faule ByteString zum Schreiben verwendet, liest die ganze Datei in den Speicher, aber der Haufen wächst fast sofort auf 70megs (die Größe der Datei) innerhalb eines Bruchteils einer Sekunde und bleibt dann konstant Datei wird verarbeitet.

Egal, wie kompliziert Ihr Filter ist, wenn Ihr Programm Gigabytes von Heap wächst, ist die Implementierung kaputt, und Sie müssen es beheben, bevor Sie sich Sorgen über Leistung, parallel oder anderweitig machen.