git.iko.soy/distributed-process
on master help sign in
Very simple worker/master example
9420187Edsko de Vries
network-transport-2/tests/sumeuler/SumEulerMaster.hsnew file+44
compare with master
unifiedsplit
@@ -0,0 +1,44 @@
1+import System.Environment (getArgs)
2+import Network.Transport
3+import Network.Transport.TCP (createTransport)
4+import Control.Concurrent (forkIO)
5+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
6+import Control.Monad (forM, forM_, replicateM_)
7+import qualified Data.ByteString as BS (concat)
8+import qualified Data.ByteString.Char8 as BSC (pack, unpack)
9+import Control.Monad.Trans.Writer (execWriterT, tell)
10+import Control.Monad.IO.Class (liftIO)
11+
12+master :: MVar () -> EndPoint -> [String] -> IO ()
13+master done endpoint workers = do
14+ conns <- forM workers $ \worker -> do
15+ Right conn <- connect endpoint (EndPointAddress $ BSC.pack worker) ReliableOrdered
16+ return conn
17+ -- Send out requests
18+ forM_ conns $ \conn -> do
19+ send conn [BSC.pack $ show 5300]
20+ close conn
21+ -- Print all replies
22+ replies <- execWriterT $ replicateM_ (length workers * 3) $ do
23+ event <- liftIO $ receive endpoint
24+ case event of
25+ Received _ msg ->
26+ tell [read . BSC.unpack . BS.concat $ msg]
27+ _ ->
28+ return ()
29+ putStrLn $ "Replies: " ++ show (replies :: [Int])
30+ putMVar done ()
31+
32+main :: IO ()
33+main = do
34+ host:port:workers <- getArgs
35+ Right transport <- createTransport host port
36+ Right endpoint <- newEndPoint transport
37+ masterDone <- newEmptyMVar
38+
39+ putStrLn $ "Master using workers " ++ show workers
40+
41+ forkIO $ master masterDone endpoint workers
42+
43+ takeMVar masterDone
44+
network-transport-2/tests/sumeuler/SumEulerWorker.hsnew file+52
@@ -0,0 +1,52 @@
1+import System.Environment (getArgs)
2+import Network.Transport
3+import Network.Transport.TCP (createTransport)
4+import qualified Data.ByteString.Char8 as BSC (putStrLn, pack, unpack)
5+import qualified Data.ByteString as BS (concat)
6+import Control.Concurrent (forkIO)
7+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
8+import System.IO (hFlush, stdout, stderr, hPutStrLn)
9+
10+mkList :: Int -> [Int]
11+mkList n = [1 .. n - 1]
12+
13+relPrime :: Int -> Int -> Bool
14+relPrime x y = gcd x y == 1
15+
16+euler :: Int -> Int
17+euler n = length (filter (relPrime n) (mkList n))
18+
19+sumEuler :: Int -> Int
20+sumEuler = sum . (map euler) . mkList
21+
22+worker :: String -> MVar () -> EndPoint -> IO ()
23+worker id done endpoint = do
24+ ConnectionOpened _ _ theirAddr <- receive endpoint
25+ Right replyChan <- connect endpoint theirAddr ReliableOrdered
26+ go replyChan
27+ where
28+ go replyChan = do
29+ event <- receive endpoint
30+ case event of
31+ ConnectionClosed _ -> do
32+ close replyChan
33+ putMVar done ()
34+ Received _ msg -> do
35+ let i :: Int
36+ i = read . BSC.unpack . BS.concat $ msg
37+ send replyChan [BSC.pack . show $ sumEuler i]
38+ go replyChan
39+
40+main :: IO ()
41+main = do
42+ (id:host:port:_) <- getArgs
43+ Right transport <- createTransport host port
44+ Right endpoint <- newEndPoint transport
45+ workerDone <- newEmptyMVar
46+
47+ BSC.putStrLn (endPointAddressToByteString (address endpoint))
48+ hFlush stdout
49+
50+ forkIO $ worker id workerDone endpoint
51+
52+ takeMVar workerDone
network-transport-2/tests/sumeuler/sumeuler.shnew file+20
@@ -0,0 +1,20 @@
1+#/bin/bash
2+
3+rm -f workers
4+killall -9 SumEulerWorker
5+
6+ghc -O2 -i../src -XScopedTypeVariables SumEulerWorker
7+ghc -O2 -i../src -XScopedTypeVariables SumEulerMaster
8+
9+./SumEulerWorker 1 127.0.0.1 8080 >> workers &
10+./SumEulerWorker 1 127.0.0.1 8081 >> workers &
11+./SumEulerWorker 1 127.0.0.1 8082 >> workers &
12+./SumEulerWorker 1 127.0.0.1 8083 >> workers &
13+./SumEulerWorker 1 127.0.0.1 8084 >> workers &
14+./SumEulerWorker 1 127.0.0.1 8085 >> workers &
15+./SumEulerWorker 1 127.0.0.1 8086 >> workers &
16+./SumEulerWorker 1 127.0.0.1 8087 >> workers &
17+
18+echo "Waiting for all workers to be ready"
19+sleep 1
20+cat workers | xargs ./SumEulerMaster 127.0.0.1 8090