par monad implementation

Author: yefrig

wc.hs

@@ -3,7 +3,7 @@
 import Control.Parallel(pseq)
 import Control.Parallel.Strategies
 import Data.Char(isAlpha, toLower)
-import Data.Map(Map, keys, fromListWith, toList, unionsWith, insert)
+import Data.Map(Map, keys, fromListWith, toList, unionWith, insert, empty)
 import qualified Data.ByteString.Lazy.Char8 as B
 import Data.List(sortBy)
 import Data.Function(on)
@@ -49,7 +49,8 @@ main = do
     case args of 
         [filename, "par"] -> do
             content <- B.readFile filename
-            print ""
+            print $ length $ pipeline 10000 content 
+
 
             -- print $ length $ withStrategy (parBuffer 100 rdeepseq) (map wcmap (chunk 10000 (map removeNonLetters $ B.words content)))
             -- print $ take 10 $ sort $ wcpar content
@@ -68,11 +69,34 @@ main = do
 wcmap :: Stream B.ByteString -> Par (Stream (B.ByteString, Int))
 wcmap = streamMap (\bs -> (bs, 1))
 
-wcreduce :: Stream (Stream (B.ByteString, Int)) -> Par (Stream (B.ByteString, Int))
-wcreduce = streamMap (streamFold insertTuple empty)
+-- wcreduce :: Stream (Stream (B.ByteString, Int)) -> Par (Stream (Map B.ByteString Int))
+-- wcreduce = streamMap ((runPar . streamFold (insertTuple) empty))
+
+wcreduce :: Stream (B.ByteString, Int) -> Par (Map B.ByteString Int)
+wcreduce = streamFold (insertTuple) empty
+
+
+finalreduce :: Stream (Map B.ByteString Int) -> Par (Map B.ByteString Int)
+finalreduce = streamFold (unionWith (+)) empty
+
+insertTuple :: Map B.ByteString Int -> (B.ByteString, Int) -> Map B.ByteString Int
+insertTuple m (k,v) = insert k v m
+
+chunk :: Int -> [a] -> [[a]]
+chunk _ [] = []
+chunk n xs = let (as,bs) = splitAt n xs in as : chunk n bs
+
+removeNonLetters :: B.ByteString -> B.ByteString
+removeNonLetters = B.filter isAlpha . B.map toLower
 
-insertTuple :: (B.ByteString, Int) -> Map B.ByteString Int -> Map B.ByteString Int
-insertTuple (k,v) = insert k v
+pipeline :: Int -> B.ByteString -> [(B.ByteString, Int)]
+pipeline n bs = runPar $ do
+    s0 <- streamFromList (chunk n (map removeNonLetters (B.words bs))) -- stream of lists
+    s1 <- streamMap (runPar . streamFromList) s0 -- make stream of streams
+    s2 <- streamMap (runPar . wcmap) s1 -- gives stream of streams for reduce
+    s3 <- streamMap (runPar . wcreduce) s2 -- stream of maps
+    s4 <- finalreduce s3
+    return $ toList s4