@@ -15,14 +15,13 @@ import Data.ByteString.Lazy (ByteString)
1515import Data.Map (Map)
1616import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
1717import Control.Arrow (second)
18−import Control.Monad (forM_, forever, replicateM)
18+import Control.Monad (forM_, replicateM, replicateM_)
1919import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
2020import qualified Control.Concurrent.Chan as Chan
2121import Control.Distributed.Process
2222import Control.Distributed.Process.Serializable (Serializable)
2323import Control.Distributed.Process.Closure
2424import Control.Distributed.Static (closureApply, staticCompose, staticApply)
25−import Control.Concurrent (threadDelay)
2625
2726--------------------------------------------------------------------------------
2827-- Definition of a MapReduce skeleton and a local implementation --
@@ -64,59 +63,52 @@ matchDict SerializableDict = match
6463sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
6564sendDict SerializableDict = send
6665
66+sdictProcessIdPair :: SerializableDict (ProcessId, ProcessId)
67+sdictProcessIdPair = SerializableDict
68+
6769mapperProcess :: forall k1 v1 k2 v2 v3.
68− SerializableDict (ProcessId, k1, v1)
70+ SerializableDict (k1, v1)
6971 -> SerializableDict [(k2, v2)]
70− -> ProcessId
72+ -> (ProcessId, ProcessId)
7173 -> MapReduce k1 v1 k2 v2 v3
7274 -> Process ()
73−mapperProcess dictIn dictOut workQueue mr = do
74− us <- getSelfPid
75− liftIO . putStrLn $ "Mapper " ++ show us ++ " started"
76− go us
75+mapperProcess dictIn dictOut (master, workQueue) mr = getSelfPid >>= go
7776 where
7877 go us = do
7978 -- Ask the queue for work
8079 send workQueue us
8180
82− liftIO . putStrLn $ "Mapper " ++ show us ++ " waiting for work"
83− -- Wait for a reply
84− mWork <- receiveWait
85− [ matchDict dictIn $ \(pid, key, val) -> return $ Just (pid, key, val)
86− , match $ \() -> return Nothing
81+ -- Wait for a reply; if there is work, do it and repeat; otherwise, exit
82+ receiveWait
83+ [ matchDict dictIn $ \(key, val) -> do
84+ sendDict dictOut master (mrMap mr key val)
85+ go us
86+ , match $ \() ->
87+ return ()
8788 ]
8889
89− -- If there is work, do it and repeat; otherwise, exit
90− case mWork of
91− Just (pid, key, val) -> do
92− sendDict dictOut pid (mrMap mr key val)
93− go us
94−
95− Nothing -> do
96− liftIO $ putStrLn "Slave exiting"
97− return ()
98−
99−remotable ['mapperProcess]
90+remotable ['mapperProcess, 'sdictProcessIdPair]
10091
10192mapperProcessClosure :: forall k1 v1 k2 v2 v3.
10293 (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
103− => Static (SerializableDict (ProcessId, k1, v1))
94+ => Static (SerializableDict (k1, v1))
10495 -> Static (SerializableDict [(k2, v2)])
10596 -> Closure (MapReduce k1 v1 k2 v2 v3)
10697 -> ProcessId
98+ -> ProcessId
10799 -> Closure (Process ())
108−mapperProcessClosure dictIn dictOut mr pid =
109− closure decoder (encode pid) `closureApply` mr
100+mapperProcessClosure dictIn dictOut mr master workQueue =
101+ closure decoder (encode (master, workQueue)) `closureApply` mr
110102 where
111103 decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
112104 decoder =
113105 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
114106 `staticCompose`
115− staticDecode sdictProcessId
107+ staticDecode $(mkStatic 'sdictProcessIdPair)
116108
117109distrMapReduce :: forall k1 k2 v1 v2 v3 a.
118110 (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
119− => Static (SerializableDict (ProcessId, k1, v1))
111+ => Static (SerializableDict (k1, v1))
120112 -> Static (SerializableDict [(k2, v2)])
121113 -> Closure (MapReduce k1 v1 k2 v2 v3)
122114 -> [NodeId]
@@ -125,8 +117,8 @@ distrMapReduce :: forall k1 k2 v1 v2 v3 a.
125117distrMapReduce dictIn dictOut mr mappers p = do
126118 mr' <- unClosure mr
127119 us <- getSelfPid
128− queue <- liftIO $ Chan.newChan
129− slavesTerminated <- liftIO $ newEmptyMVar
120+ queue <- liftIO Chan.newChan
121+ slavesTerminated <- liftIO newEmptyMVar
130122
131123 workQueue <- spawnLocal $ do
132124 let go :: Process ()
@@ -136,18 +128,18 @@ distrMapReduce dictIn dictOut mr mappers p = do
136128 Just (key, val) -> do
137129 -- As long there is work, make it available to the mappers
138130 them <- expect
139− send them (us, key, val)
131+ send them (key, val)
140132 go
141133 Nothing -> do
142134 -- Tell the mappers to terminate
143− replicateM (length mappers) $ do
135+ replicateM_ (length mappers) $ do
144136 them <- expect
145137 send them ()
146138 liftIO $ putMVar slavesTerminated ()
147139 go
148140
149141 -- Start the mappers
150− forM_ mappers $ flip spawn (mapperProcessClosure dictIn dictOut mr workQueue)
142+ forM_ mappers $ \nid -> spawn nid (mapperProcessClosure dictIn dictOut mr us workQueue)
151143
152144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
153145 iteration input = do