git.iko.soy/distributed-process
on master help sign in
Workstealing demo
df5d32eEdsko de Vries
distributed-process-demos/distributed-process-demos.cabal+14
compare with master
unifiedsplit
@@ -48,3 +48,17 @@ executable distributed-process-azure-masterslave
48 cpp-options: -DUSE_AZURE
48 cpp-options: -DUSE_AZURE
49 else
49 else
50 buildable: False
50 buildable: False
51+
52+executable distributed-process-simplelocalnet-workstealing
53+ main-is: SimpleLocalnet.hs
54+ hs-source-dirs: src/WorkStealing
55+ build-depends: base ==4.5.*,
56+ distributed-process >= 0.3.2
57+ extensions: BangPatterns, CPP, TemplateHaskell
58+ ghc-options: -Wall
59+ if flag(use-simplelocalnet)
60+ build-depends: distributed-process-simplelocalnet
61+ cpp-options: -DUSE_SIMPLELOCALNET
62+ else
63+ buildable: False
64+
distributed-process-demos/src/WorkStealing/SimpleLocalnet.hsnew file+22
@@ -0,0 +1,22 @@
1+import System.Environment (getArgs)
2+import Control.Distributed.Process
3+import Control.Distributed.Process.Node (initRemoteTable)
4+import Control.Distributed.Process.Backend.SimpleLocalnet
5+import qualified WorkStealing
6+
7+rtable :: RemoteTable
8+rtable = WorkStealing.__remoteTable initRemoteTable
9+
10+main :: IO ()
11+main = do
12+ args <- getArgs
13+
14+ case args of
15+ ["master", host, port, n] -> do
16+ backend <- initializeBackend host port rtable
17+ startMaster backend $ \slaves -> do
18+ result <- WorkStealing.master (read n) slaves
19+ liftIO $ print result
20+ ["slave", host, port] -> do
21+ backend <- initializeBackend host port rtable
22+ startSlave backend
distributed-process-demos/src/WorkStealing/WorkStealing.hsnew file+57
@@ -0,0 +1,57 @@
1+module WorkStealing where
2+
3+import Control.Monad
4+import Control.Distributed.Process
5+import Control.Distributed.Process.Closure
6+
7+fib :: Integer -> Integer
8+fib = go (0, 1)
9+ where
10+ go (!a, !b) !n | n == 0 = a
11+ | otherwise = go (b, a + b) (n - 1)
12+
13+slave :: ProcessId -> Process ()
14+slave workQueue = do
15+ us <- getSelfPid
16+ go us
17+ where
18+ go us = do
19+ -- Ask the queue for work
20+ send workQueue us
21+
22+ -- Wait for a reply
23+ mWork <- receiveWait
24+ [ match $ \(pid, n) -> return $ Just (pid, n)
25+ , match $ \() -> return Nothing
26+ ]
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+
33+remotable ['slave]
34+
35+master :: Integer -> [NodeId] -> Process Integer
36+master n slaves = do
37+ us <- getSelfPid
38+
39+ workQueue <- spawnLocal $ do
40+ -- As long as there is work, return the next Fib to compute
41+ forM_ [1 .. n] $ \m -> do
42+ them <- expect
43+ send them (us, m)
44+
45+ -- After that, just report that the work is done
46+ forever $ do
47+ pid <- expect
48+ send pid ()
49+
50+ -- Start processes on the slaves that compute Fibonacci numbers
51+ forM_ slaves $ flip spawn ($(mkClosure 'slave) workQueue)
52+
53+ -- Wait for the result
54+ partials <- replicateM (fromInteger n) (expect :: Process Integer)
55+
56+ -- And return the sum
57+ return (sum partials)