@@ -17,7 +17,6 @@ import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
17import Control.Arrow (second)
17import Control.Arrow (second)
18import Control.Monad (forM_, replicateM, replicateM_)
18import Control.Monad (forM_, replicateM, replicateM_)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
20−import qualified Control.Concurrent.Chan as Chan
21import Control.Distributed.Process
20import Control.Distributed.Process
22import Control.Distributed.Process.Serializable (Serializable)
21import Control.Distributed.Process.Serializable (Serializable)
23import Control.Distributed.Process.Closure
22import Control.Distributed.Process.Closure
@@ -117,13 +116,12 @@ distrMapReduce :: forall k1 k2 v1 v2 v3 a.
117distrMapReduce dictIn dictOut mr mappers p = do
116distrMapReduce dictIn dictOut mr mappers p = do
118 mr' <- unClosure mr
117 mr' <- unClosure mr
119 us <- getSelfPid
118 us <- getSelfPid
120− queue <- liftIO Chan.newChan
121 slavesTerminated <- liftIO newEmptyMVar
119 slavesTerminated <- liftIO newEmptyMVar
123− workQueue <- spawnLocal $ do
121+ workQueue <- spawnChannelLocal $ \queue -> do
124 let go :: Process ()
122 let go :: Process ()
126− mWork <- liftIO $ Chan.readChan queue
124+ mWork <- receiveChan queue
127 case mWork of
125 case mWork of
128 Just (key, val) -> do
126 Just (key, val) -> do
129 -- As long there is work, make it available to the mappers
127 -- As long there is work, make it available to the mappers
@@ -139,12 +137,14 @@ distrMapReduce dictIn dictOut mr mappers p = do
141 -- Start the mappers
139 -- Start the mappers
140+ let workQueuePid = sendPortProcessId (sendPortId workQueue)
141+ forM_ mappers $ \nid ->
142− forM_ mappers $ \nid -> spawn nid (mapperProcessClosure dictIn dictOut mr us workQueue)
142+ spawn nid (mapperProcessClosure dictIn dictOut mr us workQueuePid)
144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
145 iteration input = do
145 iteration input = do
146 -- Make work available to the mappers
146 -- Make work available to the mappers
147− liftIO $ mapM_ (Chan.writeChan queue . Just) (Map.toList input)
147+ mapM_ (sendChan workQueue . Just) (Map.toList input)
149 -- Wait for the partial results
149 -- Wait for the partial results
150 partials <- replicateM (Map.size input) expect
150 partials <- replicateM (Map.size input) expect
@@ -155,8 +155,7 @@ distrMapReduce dictIn dictOut mr mappers p = do
155 result <- p iteration
155 result <- p iteration
157 -- Terminate the wrappers
157 -- Terminate the wrappers
158− liftIO $ do
158+ sendChan workQueue Nothing
159− Chan.writeChan queue Nothing
160− takeMVar slavesTerminated
159+ liftIO $ takeMVar slavesTerminated
162 return result
161 return result