@@ -0,0 +1,155 @@
1+{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable, GADTs #-}
2+module MapReduce
3+ ( -- * Map-reduce skeleton and implementation
4+ MapReduce(..)
5+ , localMapReduce
6+ , distrMapReduce
7+ , __remoteTable
8+ -- * Re-exports from Data.Map
9+ , Map
10+ ) where
11+
12+import Data.Typeable (Typeable)
13+import Data.Binary (encode)
14+import Data.ByteString.Lazy (ByteString)
15+import Data.Map (Map)
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)
25+
26+--------------------------------------------------------------------------------
27+-- Definition of a MapReduce skeleton and a local implementation --
28+--------------------------------------------------------------------------------
29+
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)
35+
36+-- | Local (non-distributed) implementation of the map-reduce algorithm
37+--
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
43+ -> Map k1 v1
44+ -> Map k2 v2
45+localMapReduce mr = reducePerKey mr . groupByKey . mapPerKey mr
46+
47+reducePerKey :: MapReduce k1 v1 k2 v2 -> Map k2 [v2] -> Map k2 v2
48+reducePerKey mr = Map.mapWithKey (mrReduce mr)
49+
50+groupByKey :: Ord k2 => [(k2, v2)] -> Map k2 [v2]
51+groupByKey = Map.fromListWith (++) . map (second return)
52+
53+mapPerKey :: MapReduce k1 v1 k2 v2 -> Map k1 v1 -> [(k2, v2)]
54+mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
55+
56+--------------------------------------------------------------------------------
57+-- Simple distributed implementation --
58+--------------------------------------------------------------------------------
59+
60+matchDict :: forall a b. SerializableDict a -> (a -> Process b) -> Match b
61+matchDict SerializableDict = match
62+
63+sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
64+sendDict SerializableDict = send
65+
66+mapperProcess :: forall k1 v1 k2 v2.
67+ SerializableDict (ProcessId, k1, v1)
68+ -> SerializableDict [(k2, v2)]
69+ -> ProcessId
70+ -> MapReduce k1 v1 k2 v2
71+ -> Process ()
72+mapperProcess dictIn dictOut workQueue mr = do
73+ us <- getSelfPid
74+ liftIO . putStrLn $ "Mapper " ++ show us ++ " started"
75+ go us
76+ where
77+ go us = do
78+ -- Ask the queue for work
79+ send workQueue us
80+
81+ liftIO . putStrLn $ "Mapper " ++ show us ++ " waiting for work"
82+ -- Wait for a reply
83+ mWork <- receiveWait
84+ [ matchDict dictIn $ \(pid, key, val) -> return $ Just (pid, key, val)
85+ , match $ \() -> return Nothing
86+ ]
87+
88+ -- If there is work, do it and repeat; otherwise, exit
89+ case mWork of
90+ Just (pid, key, val) -> do
91+ sendDict dictOut pid (mrMap mr key val)
92+ go us
93+
94+ Nothing -> do
95+ liftIO $ putStrLn "Slave exiting"
96+ return ()
97+
98+remotable ['mapperProcess]
99+
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)
105+ -> ProcessId
106+ -> Closure (Process ())
107+mapperProcessClosure dictIn dictOut mr pid =
108+ closure decoder (encode pid) `closureApply` mr
109+ where
110+ decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 -> Process ())
111+ decoder =
112+ ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
113+ `staticCompose`
114+ staticDecode sdictProcessId
115+
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)
121+ -> [NodeId]
122+ -> Map k1 v1
123+ -> Process (Map k2 v2)
124+distrMapReduce dictIn dictOut mr mappers input = do
125+ us <- getSelfPid
126+ slavesTerminated <- liftIO $ newEmptyMVar
127+
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
131+ them <- expect
132+ send them (us, key, val)
133+
134+ -- After that, just report that the work is done
135+ replicateM (length mappers) $ do
136+ pid <- expect
137+ send pid ()
138+
139+ liftIO $ putMVar slavesTerminated ()
140+
141+ -- Start the mappers
142+ liftIO $ print mappers
143+ forM_ mappers $ flip spawn (mapperProcessClosure dictIn dictOut mr workQueue)
144+
145+ liftIO $ threadDelay 1000000
146+
147+ -- Wait for the result from the mappers
148+ partials <- replicateM (Map.size input) expect
149+
150+ -- Wait for the slaves to terminate
151+ liftIO $ takeMVar slavesTerminated
152+
153+ -- We reduce on this node
154+ mr' <- unClosure mr
155+ return (reducePerKey mr' . groupByKey . concat $ partials)