@@ -3,23 +3,17 @@ module MapReduce
33 ( -- * Map-reduce skeleton and implementation
44 MapReduce(..)
55 , localMapReduce
6− , distrMapReduce
7− , __remoteTable
6+ -- * Map-reduce algorithmic components
7+ , reducePerKey
8+ , groupByKey
89 -- * Re-exports from Data.Map
910 , Map
1011 ) where
1112
1213import Data.Typeable (Typeable)
13−import Data.Binary (encode)
14−import Data.ByteString.Lazy (ByteString)
1514import Data.Map (Map)
16−import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
15+import qualified Data.Map as Map (mapWithKey, fromListWith, toList)
1716import Control.Arrow (second)
18−import Control.Monad (forM_, replicateM, replicateM_)
19−import Control.Distributed.Process
20−import Control.Distributed.Process.Serializable (Serializable)
21−import Control.Distributed.Process.Closure
22−import Control.Distributed.Static (closureApply, staticCompose, staticApply)
2317
2418--------------------------------------------------------------------------------
2519-- Definition of a MapReduce skeleton and a local implementation --
@@ -50,112 +44,3 @@ groupByKey = Map.fromListWith (++) . map (second return)
5044
5145mapPerKey :: MapReduce k1 v1 k2 v2 v3 -> Map k1 v1 -> [(k2, v2)]
5246mapPerKey mr = concatMap (uncurry (mrMap mr)) . Map.toList
53−
54−--------------------------------------------------------------------------------
55−-- Simple distributed implementation --
56−--------------------------------------------------------------------------------
57−
58−matchDict :: forall a b. SerializableDict a -> (a -> Process b) -> Match b
59−matchDict SerializableDict = match
60−
61−sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
62−sendDict SerializableDict = send
63−
64−sdictProcessIdPair :: SerializableDict (ProcessId, ProcessId)
65−sdictProcessIdPair = SerializableDict
66−
67−mapperProcess :: forall k1 v1 k2 v2 v3.
68− SerializableDict (k1, v1)
69− -> SerializableDict [(k2, v2)]
70− -> (ProcessId, ProcessId)
71− -> MapReduce k1 v1 k2 v2 v3
72− -> Process ()
73−mapperProcess dictIn dictOut (master, workQueue) mr = getSelfPid >>= go
74− where
75− go us = do
76− -- Ask the queue for work
77− send workQueue us
78−
79− -- Wait for a reply; if there is work, do it and repeat; otherwise, exit
80− receiveWait
81− [ matchDict dictIn $ \(key, val) -> do
82− sendDict dictOut master (mrMap mr key val)
83− go us
84− , match $ \() ->
85− return ()
86− ]
87−
88−remotable ['mapperProcess, 'sdictProcessIdPair]
89−
90−mapperProcessClosure :: forall k1 v1 k2 v2 v3.
91− (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
92− => Static (SerializableDict (k1, v1))
93− -> Static (SerializableDict [(k2, v2)])
94− -> Closure (MapReduce k1 v1 k2 v2 v3)
95− -> ProcessId
96− -> ProcessId
97− -> Closure (Process ())
98−mapperProcessClosure dictIn dictOut mr master workQueue =
99− closure decoder (encode (master, workQueue)) `closureApply` mr
100− where
101− decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
102− decoder =
103− ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
104− `staticCompose`
105− staticDecode $(mkStatic 'sdictProcessIdPair)
106−
107−distrMapReduce :: forall k1 k2 v1 v2 v3 a.
108− (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
109− => Static (SerializableDict (k1, v1))
110− -> Static (SerializableDict [(k2, v2)])
111− -> Closure (MapReduce k1 v1 k2 v2 v3)
112− -> [NodeId]
113− -> ((Map k1 v1 -> Process (Map k2 v3)) -> Process a)
114− -> Process a
115−distrMapReduce dictIn dictOut mr mappers p = do
116− mr' <- unClosure mr
117− master <- getSelfPid
118−
119− workQueue <- spawnChannelLocal $ \queue -> do
120− let go :: Process ()
121− go = do
122− mWork <- receiveChan queue
123− case mWork of
124− Just (key, val) -> do
125− -- As long there is work, make it available to the mappers
126− them <- expect
127− send them (key, val)
128− go
129− Nothing -> do
130− -- Tell the mappers to terminate
131− replicateM_ (length mappers) $ do
132− them <- expect
133− send them ()
134−
135− -- Tell the master that the slaves are terminated
136− send master ()
137− go
138−
139− -- Start the mappers
140− let workQueuePid = sendPortProcessId (sendPortId workQueue)
141− forM_ mappers $ \nid ->
142− spawn nid (mapperProcessClosure dictIn dictOut mr master workQueuePid)
143−
144− let iteration :: Map k1 v1 -> Process (Map k2 v3)
145− iteration input = do
146− -- Make work available to the mappers
147− mapM_ (sendChan workQueue . Just) (Map.toList input)
148−
149− -- Wait for the partial results
150− partials <- replicateM (Map.size input) expect
151−
152− -- We reduce on this node
153− return (reducePerKey mr' . groupByKey . concat $ partials)
154−
155− result <- p iteration
156−
157− -- Terminate the wrappers
158− sendChan workQueue Nothing
159− expect :: Process ()
160−
161− return result