@@ -16,7 +16,6 @@ import Data.Map (Map)
1616import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
1717import Control.Arrow (second)
1818import Control.Monad (forM_, replicateM, replicateM_)
19−import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
2019import Control.Distributed.Process
2120import Control.Distributed.Process.Serializable (Serializable)
2221import Control.Distributed.Process.Closure
@@ -114,10 +113,9 @@ distrMapReduce :: forall k1 k2 v1 v2 v3 a.
114113 -> ((Map k1 v1 -> Process (Map k2 v3)) -> Process a)
115114 -> Process a
116115distrMapReduce dictIn dictOut mr mappers p = do
117− mr' <- unClosure mr
118− us <- getSelfPid
119− slavesTerminated <- liftIO newEmptyMVar
120−
116+ mr' <- unClosure mr
117+ master <- getSelfPid
118+
121119 workQueue <- spawnChannelLocal $ \queue -> do
122120 let go :: Process ()
123121 go = do
@@ -133,13 +131,15 @@ distrMapReduce dictIn dictOut mr mappers p = do
133131 replicateM_ (length mappers) $ do
134132 them <- expect
135133 send them ()
136− liftIO $ putMVar slavesTerminated ()
134+
135+ -- Tell the master that the slaves are terminated
136+ send master ()
137137 go
138138
139139 -- Start the mappers
140140 let workQueuePid = sendPortProcessId (sendPortId workQueue)
141141 forM_ mappers $ \nid ->
142− spawn nid (mapperProcessClosure dictIn dictOut mr us workQueuePid)
142+ spawn nid (mapperProcessClosure dictIn dictOut mr master workQueuePid)
143143
144144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
145145 iteration input = do
@@ -156,6 +156,6 @@ distrMapReduce dictIn dictOut mr mappers p = do
156156
157157 -- Terminate the wrappers
158158 sendChan workQueue Nothing
159− liftIO $ takeMVar slavesTerminated
159+ expect :: Process ()
160160
161161 return result