git.iko.soy/distributed-process
on master help sign in
Clean up the code
bd36d06Edsko de Vries
distributed-process-demos/distributed-process-demos.cabal+1−1
compare with master
unifiedsplit
@@ -93,7 +93,7 @@ executable distributed-process-simplelocalnet-mapreduce
93 hs-source-dirs: src/MapReduce
93 hs-source-dirs: src/MapReduce
94 build-depends: base ==4.5.*,
94 build-depends: base ==4.5.*,
95 distributed-process >= 0.3.2,
95 distributed-process >= 0.3.2,
96− distributed-static >= 0.2 && < 0.3,
96+ distributed-static >= 0.2.0 && < 0.3,
97 containers >= 0.4 && < 0.6,
97 containers >= 0.4 && < 0.6,
98 bytestring >= 0.9 && < 0.11,
98 bytestring >= 0.9 && < 0.11,
99 binary >= 0.5 && < 0.6,
99 binary >= 0.5 && < 0.6,
distributed-process-demos/src/MapReduce/CountWords.hs+1−1
@@ -23,7 +23,7 @@ countWords = MapReduce {
23localCountWords :: Map FilePath Document -> Map Word Frequency
23localCountWords :: Map FilePath Document -> Map Word Frequency
24localCountWords = localMapReduce countWords
24localCountWords = localMapReduce countWords
25
25
26−dictIn :: SerializableDict (ProcessId, FilePath, Document)
26+dictIn :: SerializableDict (FilePath, Document)
27dictIn = SerializableDict
27dictIn = SerializableDict
28
28
29dictOut :: SerializableDict [(Word, Frequency)]
29dictOut :: SerializableDict [(Word, Frequency)]
distributed-process-demos/src/MapReduce/KMeans.hs+3−3
@@ -59,7 +59,7 @@ localKMeans points cs iterations = go (iterations - 1)
59 trivialSegmentation :: [Cluster] -> Map (Int, Int) [Cluster]
59 trivialSegmentation :: [Cluster] -> Map (Int, Int) [Cluster]
60 trivialSegmentation cs' = Map.fromList [(bounds points, cs')]
60 trivialSegmentation cs' = Map.fromList [(bounds points, cs')]
61
61
62−dictIn :: SerializableDict (ProcessId, (Int, Int), [Cluster])
62+dictIn :: SerializableDict ((Int, Int), [Cluster])
63dictIn = SerializableDict
63dictIn = SerializableDict
64
64
65dictOut :: SerializableDict [(Cluster, Point)]
65dictOut :: SerializableDict [(Cluster, Point)]
@@ -82,7 +82,7 @@ distrKMeans points cs mappers iterations =
82 go :: Int
82 go :: Int
83 -> (Map (Int, Int) [Cluster] -> Process (Map Cluster ([Point], Point)))
83 -> (Map (Int, Int) [Cluster] -> Process (Map Cluster ([Point], Point)))
84 -> Process (Map Cluster ([Point], Point))
84 -> Process (Map Cluster ([Point], Point))
85− go 0 iteration = do
85+ go 0 iteration =
86 iteration (Map.fromList $ map (, cs) segments)
86 iteration (Map.fromList $ map (, cs) segments)
87 go n iteration = do
87 go n iteration = do
88 clusters <- go (n - 1) iteration
88 clusters <- go (n - 1) iteration
@@ -120,7 +120,7 @@ createGnuPlot clusters h =
120
120
121 flatten :: [(Float, (KMeans.Cluster, ([KMeans.Point], KMeans.Point)))]
121 flatten :: [(Float, (KMeans.Cluster, ([KMeans.Point], KMeans.Point)))]
122 -> [(Double, Double, Float)]
122 -> [(Double, Double, Float)]
123− flatten = concat . map (\(color, (_, (points, _))) -> map (\(x, y) -> (x, y, color)) points)
123+ flatten = concatMap (\(color, (_, (points, _))) -> map (\(x, y) -> (x, y, color)) points)
124
124
125 colors :: [Float]
125 colors :: [Float]
126 colors = [0, 1 / fromIntegral (Map.size clusters) .. 1]
126 colors = [0, 1 / fromIntegral (Map.size clusters) .. 1]
distributed-process-demos/src/MapReduce/MapReduce.hs+26−34
@@ -15,14 +15,13 @@ import Data.ByteString.Lazy (ByteString)
15import Data.Map (Map)
15import Data.Map (Map)
16import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
16import qualified Data.Map as Map (mapWithKey, fromListWith, toList, size)
17import Control.Arrow (second)
17import Control.Arrow (second)
18−import Control.Monad (forM_, forever, replicateM)
18+import Control.Monad (forM_, replicateM, replicateM_)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
19import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
20import qualified Control.Concurrent.Chan as Chan
20import qualified Control.Concurrent.Chan as Chan
21import Control.Distributed.Process
21import Control.Distributed.Process
22import Control.Distributed.Process.Serializable (Serializable)
22import Control.Distributed.Process.Serializable (Serializable)
23import Control.Distributed.Process.Closure
23import Control.Distributed.Process.Closure
24import Control.Distributed.Static (closureApply, staticCompose, staticApply)
24import Control.Distributed.Static (closureApply, staticCompose, staticApply)
25−import Control.Concurrent (threadDelay)
26
25
27--------------------------------------------------------------------------------
26--------------------------------------------------------------------------------
28-- Definition of a MapReduce skeleton and a local implementation --
27-- Definition of a MapReduce skeleton and a local implementation --
@@ -64,59 +63,52 @@ matchDict SerializableDict = match
64sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
63sendDict :: forall a. SerializableDict a -> ProcessId -> a -> Process ()
65sendDict SerializableDict = send
64sendDict SerializableDict = send
66
65
66+sdictProcessIdPair :: SerializableDict (ProcessId, ProcessId)
67+sdictProcessIdPair = SerializableDict
68+
67mapperProcess :: forall k1 v1 k2 v2 v3.
69mapperProcess :: forall k1 v1 k2 v2 v3.
68− SerializableDict (ProcessId, k1, v1)
70+ SerializableDict (k1, v1)
69 -> SerializableDict [(k2, v2)]
71 -> SerializableDict [(k2, v2)]
70− -> ProcessId
72+ -> (ProcessId, ProcessId)
71 -> MapReduce k1 v1 k2 v2 v3
73 -> MapReduce k1 v1 k2 v2 v3
72 -> Process ()
74 -> Process ()
73−mapperProcess dictIn dictOut workQueue mr = do
75+mapperProcess dictIn dictOut (master, workQueue) mr = getSelfPid >>= go
74− us <- getSelfPid
75− liftIO . putStrLn $ "Mapper " ++ show us ++ " started"
76− go us
77 where
76 where
78 go us = do
77 go us = do
79 -- Ask the queue for work
78 -- Ask the queue for work
80 send workQueue us
79 send workQueue us
81
80
82− liftIO . putStrLn $ "Mapper " ++ show us ++ " waiting for work"
83− -- Wait for a reply
81+ -- Wait for a reply; if there is work, do it and repeat; otherwise, exit
84− mWork <- receiveWait
82+ receiveWait
85− [ matchDict dictIn $ \(pid, key, val) -> return $ Just (pid, key, val)
83+ [ matchDict dictIn $ \(key, val) -> do
84+ sendDict dictOut master (mrMap mr key val)
85+ go us
86+ , match $ \() ->
86− , match $ \() -> return Nothing
87+ return ()
87 ]
88 ]
88
89
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]
100
91
101mapperProcessClosure :: forall k1 v1 k2 v2 v3.
92mapperProcessClosure :: forall k1 v1 k2 v2 v3.
102 (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
93 (Typeable k1, Typeable v1, Typeable k2, Typeable v2, Typeable v3)
103− => Static (SerializableDict (ProcessId, k1, v1))
94+ => Static (SerializableDict (k1, v1))
104 -> Static (SerializableDict [(k2, v2)])
95 -> Static (SerializableDict [(k2, v2)])
105 -> Closure (MapReduce k1 v1 k2 v2 v3)
96 -> Closure (MapReduce k1 v1 k2 v2 v3)
106 -> ProcessId
97 -> ProcessId
98+ -> ProcessId
107 -> Closure (Process ())
99 -> Closure (Process ())
108−mapperProcessClosure dictIn dictOut mr pid =
100+mapperProcessClosure dictIn dictOut mr master workQueue =
109− closure decoder (encode pid) `closureApply` mr
101+ closure decoder (encode (master, workQueue)) `closureApply` mr
110 where
102 where
111 decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
103 decoder :: Static (ByteString -> MapReduce k1 v1 k2 v2 v3 -> Process ())
112 decoder =
104 decoder =
113 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
105 ($(mkStatic 'mapperProcess) `staticApply` dictIn `staticApply` dictOut)
114 `staticCompose`
106 `staticCompose`
115− staticDecode sdictProcessId
107+ staticDecode $(mkStatic 'sdictProcessIdPair)
116
108
117distrMapReduce :: forall k1 k2 v1 v2 v3 a.
109distrMapReduce :: forall k1 k2 v1 v2 v3 a.
118 (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
110 (Serializable k1, Serializable v1, Serializable k2, Serializable v2, Serializable v3, Ord k2)
119− => Static (SerializableDict (ProcessId, k1, v1))
111+ => Static (SerializableDict (k1, v1))
120 -> Static (SerializableDict [(k2, v2)])
112 -> Static (SerializableDict [(k2, v2)])
121 -> Closure (MapReduce k1 v1 k2 v2 v3)
113 -> Closure (MapReduce k1 v1 k2 v2 v3)
122 -> [NodeId]
114 -> [NodeId]
@@ -125,8 +117,8 @@ distrMapReduce :: forall k1 k2 v1 v2 v3 a.
125distrMapReduce dictIn dictOut mr mappers p = do
117distrMapReduce dictIn dictOut mr mappers p = do
126 mr' <- unClosure mr
118 mr' <- unClosure mr
127 us <- getSelfPid
119 us <- getSelfPid
128− queue <- liftIO $ Chan.newChan
120+ queue <- liftIO Chan.newChan
129− slavesTerminated <- liftIO $ newEmptyMVar
121+ slavesTerminated <- liftIO newEmptyMVar
130
122
131 workQueue <- spawnLocal $ do
123 workQueue <- spawnLocal $ do
132 let go :: Process ()
124 let go :: Process ()
@@ -136,18 +128,18 @@ distrMapReduce dictIn dictOut mr mappers p = do
136 Just (key, val) -> do
128 Just (key, val) -> do
137 -- As long there is work, make it available to the mappers
129 -- As long there is work, make it available to the mappers
138 them <- expect
130 them <- expect
139− send them (us, key, val)
131+ send them (key, val)
140 go
132 go
141 Nothing -> do
133 Nothing -> do
142 -- Tell the mappers to terminate
134 -- Tell the mappers to terminate
143− replicateM (length mappers) $ do
135+ replicateM_ (length mappers) $ do
144 them <- expect
136 them <- expect
145 send them ()
137 send them ()
146 liftIO $ putMVar slavesTerminated ()
138 liftIO $ putMVar slavesTerminated ()
147 go
139 go
148
140
149 -- Start the mappers
141 -- 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)
151
143
152 let iteration :: Map k1 v1 -> Process (Map k2 v3)
144 let iteration :: Map k1 v1 -> Process (Map k2 v3)
153 iteration input = do
145 iteration input = do
distributed-process-demos/src/MasterSlave/MasterSlave.hs+1−1
@@ -24,7 +24,7 @@ master n slaves = do
24 spawn them ($(mkClosure 'slave) (us, m))
24 spawn them ($(mkClosure 'slave) (us, m))
25
25
26 -- Wait for the result
26 -- Wait for the result
27− partials <- replicateM (fromInteger n) (expect :: Process Integer)
27+ partials <- replicateM (fromIntegral n) (expect :: Process Integer)
28
28
29 -- And return the sum
29 -- And return the sum
30 return (sum partials)
30 return (sum partials)
distributed-process-demos/src/WorkPushing/WorkPushing.hs+7−8
@@ -10,10 +10,10 @@ fib = go (0, 1)
10 go (!a, !b) !n | n == 0 = a
10 go (!a, !b) !n | n == 0 = a
11 | otherwise = go (b, a + b) (n - 1)
11 | otherwise = go (b, a + b) (n - 1)
12
12
13−slave :: () -> Process ()
13+slave :: ProcessId -> Process ()
14−slave () = forever $ do
14+slave them = forever $ do
15− (pid, n) <- expect
15+ n <- expect
16− send pid (fib n)
16+ send them (fib n)
17
17
18remotable ['slave]
18remotable ['slave]
19
19
@@ -22,14 +22,13 @@ master n slaves = do
22 us <- getSelfPid
22 us <- getSelfPid
23
23
24 -- Start processes on the slaves that compute Fibonacci numbers
24 -- Start processes on the slaves that compute Fibonacci numbers
25− slaveProcesses <- forM slaves $ flip spawn ($(mkClosure 'slave) ())
25+ slaveProcesses <- forM slaves $ \nid -> spawn nid ($(mkClosure 'slave) us)
26
26
27 -- Distribute 1 .. n amongst the slave processes
27 -- Distribute 1 .. n amongst the slave processes
28− forM_ (zip [1 .. n] (cycle slaveProcesses)) $ \(m, them) ->
28+ forM_ (zip [1 .. n] (cycle slaveProcesses)) $ \(m, them) -> send them m
29− send them (us, m)
30
29
31 -- Wait for the result
30 -- Wait for the result
32− partials <- replicateM (fromInteger n) (expect :: Process Integer)
31+ partials <- replicateM (fromIntegral n) (expect :: Process Integer)
33
32
34 -- And return the sum
33 -- And return the sum
35 return (sum partials)
34 return (sum partials)
distributed-process-demos/src/WorkStealing/WorkStealing.hs+9−14
@@ -10,8 +10,8 @@ fib = go (0, 1)
10 go (!a, !b) !n | n == 0 = a
10 go (!a, !b) !n | n == 0 = a
11 | otherwise = go (b, a + b) (n - 1)
11 | otherwise = go (b, a + b) (n - 1)
12
12
13−slave :: ProcessId -> Process ()
13+slave :: (ProcessId, ProcessId) -> Process ()
14−slave workQueue = do
14+slave (master, workQueue) = do
15 us <- getSelfPid
15 us <- getSelfPid
16 go us
16 go us
17 where
17 where
@@ -19,17 +19,12 @@ slave workQueue = do
19 -- Ask the queue for work
19 -- Ask the queue for work
20 send workQueue us
20 send workQueue us
21
21
22− -- Wait for a reply
22+ -- If there is work, do it, otherwise terminate
23− mWork <- receiveWait
23+ receiveWait
24− [ match $ \(pid, n) -> return $ Just (pid, n)
24+ [ match $ \n -> send master (fib n) >> go us
25− , match $ \() -> return Nothing
25+ , match $ \() -> return ()
26 ]
26 ]
27
27
28− -- If there is work, do it and repeat; otherwise, exit
29− case mWork of
30− Just (pid, n) -> send pid (fib n) >> go us
31− Nothing -> return ()
32−
33remotable ['slave]
28remotable ['slave]
34
29
35master :: Integer -> [NodeId] -> Process Integer
30master :: Integer -> [NodeId] -> Process Integer
@@ -40,7 +35,7 @@ master n slaves = do
40 -- As long as there is work, return the next Fib to compute
35 -- As long as there is work, return the next Fib to compute
41 forM_ [1 .. n] $ \m -> do
36 forM_ [1 .. n] $ \m -> do
42 them <- expect
37 them <- expect
43− send them (us, m)
38+ send them m
44
39
45 -- After that, just report that the work is done
40 -- After that, just report that the work is done
46 forever $ do
41 forever $ do
@@ -48,10 +43,10 @@ master n slaves = do
48 send pid ()
43 send pid ()
49
44
50 -- Start processes on the slaves that compute Fibonacci numbers
45 -- Start processes on the slaves that compute Fibonacci numbers
51− forM_ slaves $ flip spawn ($(mkClosure 'slave) workQueue)
46+ forM_ slaves $ \nid -> spawn nid ($(mkClosure 'slave) (us, workQueue))
52
47
53 -- Wait for the result
48 -- Wait for the result
54− partials <- replicateM (fromInteger n) (expect :: Process Integer)
49+ partials <- replicateM (fromIntegral n) (expect :: Process Integer)
55
50
56 -- And return the sum
51 -- And return the sum
57 return (sum partials)
52 return (sum partials)