@@ -0,0 +1,155 @@
1+{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable, GADTs #-}
3+ ( -- * Map-reduce skeleton and implementation
8+ -- * Re-exports from Data.Map
12+import Data.Typeable (Typeable)
13+import Data.Binary (encode)
14+import Data.ByteString.Lazy (ByteString)
16+import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
17+import Control.Arrow (second)
18+import Control.Monad (forM_, forever, replicateM)
19+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
20+import Control.Distributed.Process
21+import Control.Distributed.Process.Serializable (Serializable)
22+import Control.Distributed.Process.Closure
23+import Control.Distributed.Static (closureApply, staticCompose, staticApply)
24+import Control.Concurrent (threadDelay)
26+--------------------------------------------------------------------------------
27+-- Definition of a MapReduce skeleton and a local implementation --
28+--------------------------------------------------------------------------------
30+-- | MapReduce skeleton
31+data MapReduce k1 v1 k2 v2 = MapReduce {
32+ mrMap :: k1 -> v1 -> [(k2, v2)]
33+ , mrReduce :: k2 -> [v2] -> v2
34+ } deriving (Typeable)
36+-- | Local (non-distributed) implementation of the map-reduce algorithm
38+-- This can be regarded as the specification of map-reduce; see
39+-- /Google’s MapReduce Programming Model---Revisited/ by Ralf Laemmel
40+-- (<http://userpages.uni-koblenz.de/~laemmel/MapReduce/>).
41+localMapReduce :: forall k1 k2 v1 v2. Ord k2 =>
42+ MapReduce k1 v1 k2 v2
45+localMapReduce mr = reducePerKey mr . groupByKey . mapPerKey mr
47+reducePerKey :: MapReduce k1 v1 k2 v2 -> Map k2 [v2] -> Map k2 v2
48+reducePerKey mr = Map.mapWithKey (mrReduce mr)
50+groupByKey :: Ord k2 => [(k2, v2)] -> Map k2 [v2]
51+groupByKey = Map.fromListWith (++) . map (second return)
53+mapPerKey :: MapReduce k1 v1 k2 v2 -> Map k1 v1 -> [(k2, v2)]
54+mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
56+--------------------------------------------------------------------------------
57+-- Simple distributed implementation --
58+--------------------------------------------------------------------------------
60+matchDict :: forall a b. SerializableDict a -> (a -> Process b) -> Match b
61+matchDict SerializableDict = match
63+sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
64+sendDict SerializableDict = send
66+mapperProcess :: forall k1 v1 k2 v2.
67+ SerializableDict (ProcessId, k1, v1)
68+ -> SerializableDict [(k2, v2)]
70+ -> MapReduce k1 v1 k2 v2
72+mapperProcess dictIn dictOut workQueue mr = do
74+ liftIO . putStrLn $ "Mapper " ++ show us ++ " started"
78+ -- Ask the queue for work
81+ liftIO . putStrLn $ "Mapper " ++ show us ++ " waiting for work"
84+ [ matchDict dictIn $ \(pid, key, val) -> return $ Just (pid, key, val)
85+ , match $ \() -> return Nothing
88+ -- If there is work, do it and repeat; otherwise, exit
90+ Just (pid, key, val) -> do
91+ sendDict dictOut pid (mrMap mr key val)
95+ liftIO $ putStrLn "Slave exiting"
98+remotable ['mapperProcess]
100+mapperProcessClosure :: forall k1 v1 k2 v2.
101+ (Typeable k1, Typeable v1, Typeable k2, Typeable v2)
102+ => Static (SerializableDict (ProcessId, k1, v1))
103+ -> Static (SerializableDict [(k2, v2)])
104+ -> Closure (MapReduce k1 v1 k2 v2)
106+ -> Closure (Process ())
107+mapperProcessClosure dictIn dictOut mr pid =
108+ closure decoder (encode pid) `closureApply` mr
110+ decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 -> Process ())
112+ ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
114+ staticDecode sdictProcessId
116+distrMapReduce :: forall k1 k2 v1 v2.
117+ (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Ord k2)
118+ => Static (SerializableDict (ProcessId, k1, v1))
119+ -> Static (SerializableDict [(k2, v2)])
120+ -> Closure (MapReduce k1 v1 k2 v2)
123+ -> Process (Map k2 v2)
124+distrMapReduce dictIn dictOut mr mappers input = do
126+ slavesTerminated <- liftIO $ newEmptyMVar
128+ workQueue <- spawnLocal $ do
129+ -- As long as there is work, return the next Fib to compute
130+ forM_ (Map.toList input) $ \(key, val) -> do
132+ send them (us, key, val)
134+ -- After that, just report that the work is done
135+ replicateM (length mappers) $ do
139+ liftIO $ putMVar slavesTerminated ()
141+ -- Start the mappers
142+ liftIO $ print mappers
143+ forM_ mappers $ flip spawn (mapperProcessClosure dictIn dictOut mr workQueue)
145+ liftIO $ threadDelay 1000000
147+ -- Wait for the result from the mappers
148+ partials <- replicateM (Map.size input) expect
150+ -- Wait for the slaves to terminate
151+ liftIO $ takeMVar slavesTerminated
153+ -- We reduce on this node
155+ return (reducePerKey mr' . groupByKey . concat $ partials)