@@ -17,7 +17,6 @@ import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
1717import Control.Arrow (second)
1818import Control.Monad (forM_, replicateM, replicateM_)
1919import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
20−import qualified Control.Concurrent.Chan as Chan
2120import Control.Distributed.Process
2221import Control.Distributed.Process.Serializable (Serializable)
2322import Control.Distributed.Process.Closure
@@ -117,13 +116,12 @@ distrMapReduce :: forall k1 k2 v1 v2 v3 a.
117116distrMapReduce dictIn dictOut mr mappers p = do
118117 mr' <- unClosure mr
119118 us <- getSelfPid
120− queue <- liftIO Chan.newChan
121119 slavesTerminated <- liftIO newEmptyMVar
122120
123− workQueue <- spawnLocal $ do
121+ workQueue <- spawnChannelLocal $ \queue -> do
124122 let go :: Process ()
125123 go = do
126− mWork <- liftIO $ Chan.readChan queue
124+ mWork <- receiveChan queue
127125 case mWork of
128126 Just (key, val) -> do
129127 -- As long there is work, make it available to the mappers
@@ -139,12 +137,14 @@ distrMapReduce dictIn dictOut mr mappers p = do
139137 go
140138
141139 -- Start the mappers
142− forM_ mappers $ \nid -> spawn nid (mapperProcessClosure dictIn dictOut mr us workQueue)
140+ let workQueuePid = sendPortProcessId (sendPortId workQueue)
141+ forM_ mappers $ \nid ->
142+ spawn nid (mapperProcessClosure dictIn dictOut mr us workQueuePid)
143143
144144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
145145 iteration input = do
146146 -- Make work available to the mappers
147− liftIO $ mapM_ (Chan.writeChan queue . Just) (Map.toList input)
147+ mapM_ (sendChan workQueue . Just) (Map.toList input)
148148
149149 -- Wait for the partial results
150150 partials <- replicateM (Map.size input) expect
@@ -155,8 +155,7 @@ distrMapReduce dictIn dictOut mr mappers p = do
155155 result <- p iteration
156156
157157 -- Terminate the wrappers
158− liftIO $ do
159− Chan.writeChan queue Nothing
160− takeMVar slavesTerminated
158+ sendChan workQueue Nothing
159+ liftIO $ takeMVar slavesTerminated
161160
162161 return result