git.iko.soy/Fraudik/haskell-web-wc
on main help sign in
Add parallel words counting
3763165Blinov Ilya
README.md+3−1
compare with main
unifiedsplit
@@ -11,8 +11,10 @@
11[x] Первая версия, Proof of Concept. Просто работа с MultipartForm запросами, реализация wc из домашнего задания, тестирование через отправку этого README.
11[x] Первая версия, Proof of Concept. Просто работа с MultipartForm запросами, реализация wc из домашнего задания, тестирование через отправку этого README.
12[x] Подготовлен тестовый файл: sample.txt в 1,12 ГБ. Статистики первого wc для него: 25018508 194885810 1209339403 (строки, слова, символы)
12[x] Подготовлен тестовый файл: sample.txt в 1,12 ГБ. Статистики первого wc для него: 25018508 194885810 1209339403 (строки, слова, символы)
13[x] Реализован подсчет слов, проверена корректность
13[x] Реализован подсчет слов, проверена корректность
14+[x] Реализована параллельная обработка (интерфейс обращения к WordCount пришлось немного поменять для удобства)
14
15
15## Анализ производительности
16## Анализ производительности
16
17
171. Proof of Concept: Elapsed (wall clock) time (h:mm:ss or m:ss): 0:27.25, Maximum resident set size (kbytes): 1816
181. Proof of Concept: Elapsed (wall clock) time (h:mm:ss or m:ss): 0:27.25, Maximum resident set size (kbytes): 1816
18−2. Подсчет слов: Elapsed (wall clock) time (h:mm:ss or m:ss): 0:39.20, Maximum resident set size (kbytes): 1816
19+2. Подсчет слов: Elapsed (wall clock) time (h:mm:ss or m:ss): 0:39.20, Maximum resident set size (kbytes): 1816
20+3. С параллельной обработкой и 16 Haskell потоками: Elapsed (wall clock) time (h:mm:ss or m:ss): 0:13.07, Maximum resident set size (kbytes): 1820
app/Main.hs+3−3
@@ -14,7 +14,7 @@ import Network.Wai.Handler.Warp
14import Servant
14import Servant
15import Servant.Multipart
15import Servant.Multipart
16
16
17−import WordCount (foldWith, foldCountElements, printSelecttedOutput)
17+import WordCount (countElements, printSelecttedOutput)
18
18
19type API = MultipartForm Tmp (MultipartData Tmp) :> Post '[JSON] String
19type API = MultipartForm Tmp (MultipartData Tmp) :> Post '[JSON] String
20
20
@@ -28,8 +28,8 @@ upload :: Server API
28upload multipartData = do
28upload multipartData = do
29 case files multipartData of
29 case files multipartData of
30 (file:_) -> do
30 (file:_) -> do
31− wcStats <- liftIO $ foldWith foldCountElements $ fdPayload file
31+ output <- liftIO $ countElements $ fdPayload file
32− return $ printSelecttedOutput wcStats
32+ return $ printSelecttedOutput output
33 [] -> throwError noFileError
33 [] -> throwError noFileError
34
34
35startServer :: IO ()
35startServer :: IO ()
src/WordCount.hs+49−20
@@ -1,4 +1,4 @@
1−module WordCount (foldWith, foldCountElements, printSelecttedOutput) where
1+module WordCount (countElements, printSelecttedOutput) where
2
2
3import Streamly.Data.Fold (Fold)
3import Streamly.Data.Fold (Fold)
4import qualified Streamly.Data.Stream as Stream
4import qualified Streamly.Data.Stream as Stream
@@ -6,11 +6,18 @@ import qualified Streamly.Data.Fold as Fold
6import qualified Streamly.FileSystem.File as File
6import qualified Streamly.FileSystem.File as File
7import Data.Function ((&))
7import Data.Function ((&))
8import Data.Char (ord)
8import Data.Char (ord)
9+import Data.Word (Word8)
9import qualified Streamly.Unicode.Stream as Stream
10import qualified Streamly.Unicode.Stream as Stream
11+import Streamly.Data.Array (Array)
12+import qualified Streamly.Data.Stream.Prelude as Stream
13+import GHC.Conc (numCapabilities)
14+import qualified Streamly.Data.Array as Array
15+import Data.Char (chr)
10
16
11isSpace :: Char -> Bool
17isSpace :: Char -> Bool
12−isSpace c = uc == 0x20 || uc - 0x9 <= 4
18+isSpace c = uc == 32 || uc - 0x9 <= 4 || uc == 0xa0
19+ where
13− where uc = fromIntegral (ord c) :: Word
20+ uc = fromIntegral (ord c) :: Word
14
21
15data WCOutputs =
22data WCOutputs =
16 WCOutputs {
23 WCOutputs {
@@ -21,21 +28,43 @@ data WCOutputs =
21 }
28 }
22 deriving (Show)
29 deriving (Show)
23
30
24−count :: WCOutputs -> Char -> WCOutputs
31+processChar :: WCOutputs -> Char -> WCOutputs
25−count (WCOutputs charsAmount wordsAmount linesAmount wasSpace) char =
32+processChar (WCOutputs charsAmount wordsAmount linesAmount wasSpace) char =
26 let newLinesAmount = if char == '\n' then linesAmount + 1 else linesAmount
33 let newLinesAmount = if char == '\n' then linesAmount + 1 else linesAmount
27− (newWordsAmount, isSpaceChar) = if isSpace char then (wordsAmount, True)
34+ (newWordsAmount, isSpaceChar') = if isSpace char then (wordsAmount, True)
28− else (if wasSpace then wordsAmount + 1 else wordsAmount, False)
35+ else (if wasSpace then wordsAmount + 1 else wordsAmount, False)
29− in WCOutputs (charsAmount + 1) newWordsAmount newLinesAmount isSpaceChar
36+ in WCOutputs (charsAmount + 1) newWordsAmount newLinesAmount isSpaceChar'
30−
37+
31−foldWith :: Fold IO Char a -> String -> IO a
38+processChunk :: Array Word8 -> IO (Bool, WCOutputs)
32−foldWith f file =
39+processChunk chunk = do
33− File.read file
40+ let firstChar = Array.getIndex 0 chunk
41+ case firstChar of
42+ Just x -> do
43+ output <- countChunkElements chunk
44+ return (isSpace (chr (fromIntegral x)), output)
45+ Nothing -> return (False, WCOutputs 0 0 0 True)
46+ where
47+ countChunkElements chunk' =
48+ Array.read chunk'
34− & Stream.decodeUtf8
49+ & Stream.decodeUtf8
35− & Stream.fold f
50+ & Stream.fold (Fold.foldl' processChar (WCOutputs 0 0 0 True))
36−
51+
37−foldCountElements :: Fold IO Char WCOutputs
52+combineChunkOutputs :: (Bool, WCOutputs) -> (Bool, WCOutputs) -> (Bool, WCOutputs)
53+combineChunkOutputs (fstBeginsWithSpace, WCOutputs fstLines fstWords fstChars fstEndsWithSpace) (sndBeginsWithSpace, WCOutputs sndLines sndWords sndChars sndEndsWithSpace) =
54+ -- For correctly processing multiple following spaces and first/last words
55+ let combinedWordCount = if not fstEndsWithSpace && not sndBeginsWithSpace
56+ then fstWords + sndWords - 1
57+ else fstWords + sndWords
58+ in (fstBeginsWithSpace, WCOutputs (fstLines + sndLines) combinedWordCount (fstChars + sndChars) sndEndsWithSpace)
59+
60+foldChunks :: Fold IO (Bool, WCOutputs) (Bool, WCOutputs)
38−foldCountElements = Fold.foldl' count (WCOutputs 0 0 0 True)
61+foldChunks = Fold.foldl' combineChunkOutputs (False, WCOutputs 0 0 0 True)
39−
62+
63+countElements :: String -> IO (Bool, WCOutputs)
64+countElements file = do
65+ File.readChunks file
66+ & Stream.parMapM (Stream.maxThreads numCapabilities . Stream.ordered True) processChunk
67+ & Stream.fold foldChunks
68+
40−printSelecttedOutput :: WCOutputs -> String
69+printSelecttedOutput :: (Bool, WCOutputs) -> String
41−printSelecttedOutput output = show (lineCount output) ++ " " ++ show (wordsCount output) ++ " " ++ show (charCount output)
70+printSelecttedOutput output = show (lineCount . snd $ output) ++ " " ++ show (wordsCount . snd $ output) ++ " " ++ show (charCount . snd $ output)