git.iko.soy/distributed-process
on master help sign in
This checkin passes a simple demo test, but it still depends on
12559abRyan Newton
examples/DemoTransport2.hsnew file+142
compare with master
unifiedsplit
@@ -0,0 +1,142 @@
1+-- module DemoTransport where
2+module Main where
3+
4+import Network.Transport
5+import qualified Network.Transport.MVar
6+import Network.Transport.TCP (mkTransport, TCPConfig (..))
7+
8+import qualified Network.Transport.Pipes as Pipes
9+
10+import Control.Concurrent
11+import Control.Monad
12+import Control.Exception
13+import System.IO (stderr, hPutStrLn)
14+
15+import qualified Data.ByteString.Char8 as BS
16+
17+import Debug.Trace
18+
19+forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId
20+forkWithExceptions forkit descr action = do
21+ parent <- myThreadId
22+ forkit $
23+ Control.Exception.catch
24+ action
25+ (\ e -> do
26+ hPutStrLn stderr $ "Exception inside "++descr++": "++show e
27+ throwTo parent (e::SomeException)
28+ )
29+
30+-------------------------------------------
31+-- Example program using backend directly
32+--
33+
34+-- | Check if multiple messages can be sent on the same connection.
35+-- demo0 :: IO ()
36+demo0 mkTrans = do
37+ trans <- mkTrans
38+
39+ (sourceAddr, targetEnd) <- newConnection trans
40+
41+-- forkWithExceptions forkIO "receiver thread" $ logServer "logServer" targetEnd
42+-- forkWithExceptions forkOS "receiver thread" $ logServer "logServer" targetEnd
43+ forkIO $ logServer "logServer" targetEnd
44+ threadDelay 100000
45+
46+ sourceEnd <- connect sourceAddr
47+
48+ mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
49+ threadDelay 100000
50+
51+
52+main = do
53+ putStrLn "MVAR:"
54+ demo0 Network.Transport.MVar.mkTransport
55+ putStrLn "TCP:"
56+ demo0$ mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
57+ putStrLn "Pipes:"
58+ demo0 Pipes.mkTransport
59+
60+
61+
62+
63+{-
64+-- | Check endpoint serialization and deserialization.
65+demo1 :: IO ()
66+demo1 = do
67+ -- trans <- mkTransport
68+ trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
69+
70+ (sourceAddr, targetEnd) <- newConnection trans
71+
72+ forkIO $ logServer "logServer" targetEnd
73+ threadDelay 100000
74+
75+ sourceEnd <- connect sourceAddr
76+
77+ -- This use of Just is slightly naughty
78+ let Just sourceAddr' = deserialize trans (serialize sourceAddr)
79+ sourceEnd' <- connect sourceAddr'
80+
81+ send sourceEnd [BS.pack "hello 1"]
82+ send sourceEnd' [BS.pack "hello 2"]
83+ threadDelay 100000
84+
85+-- | Check that messages can be sent before receive is set up.
86+demo2 :: IO ()
87+demo2 = do
88+ -- trans <- mkTransport
89+ trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
90+
91+ (sourceAddr, targetEnd) <- newConnection trans
92+
93+ sourceEnd <- connect sourceAddr
94+ forkIO $ send sourceEnd [BS.pack "hello 1"]
95+ threadDelay 100000
96+
97+ forkIO $ logServer "logServer" targetEnd
98+ return ()
99+
100+-- | Check that two different transports can be created.
101+demo3 :: IO ()
102+demo3 = do
103+ trans1 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
104+ trans2 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8081"
105+
106+ (sourceAddr1, targetEnd1) <- newConnection trans1
107+ (sourceAddr2, targetEnd2) <- newConnection trans2
108+
109+ forkIO $ logServer "logServer1" targetEnd1
110+ forkIO $ logServer "logServer2" targetEnd2
111+
112+ sourceEnd1 <- connect sourceAddr1
113+ sourceEnd2 <- connect sourceAddr2
114+
115+ send sourceEnd1 [BS.pack "hello1"]
116+ send sourceEnd2 [BS.pack "hello2"]
117+
118+-- | Check that two different connections on the same transport can be created.
119+demo4 :: IO ()
120+demo4 = do
121+ trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
122+ -- trans <- Network.Transport.MVar.mkTransport
123+
124+ (sourceAddr1, targetEnd1) <- newConnection trans
125+ (sourceAddr2, targetEnd2) <- newConnection trans
126+
127+ forkIO $ logServer "logServer1" targetEnd1
128+ forkIO $ logServer "logServer2" targetEnd2
129+
130+ sourceEnd1 <- connect sourceAddr1
131+ sourceEnd2 <- connect sourceAddr2
132+
133+ send sourceEnd1 [BS.pack "hello1"]
134+ send sourceEnd2 [BS.pack "hello2"]
135+
136+-}
137+
138+logServer :: String -> TargetEnd -> IO ()
139+logServer name targetEnd = forever $ do
140+ x <- receive targetEnd
141+ trace (name ++ " rcvd: " ++ show x) $ return ()
142+
network-transport/src/Network/Transport/Pipes.hs+64−31
@@ -1,4 +1,4 @@
1−{-# LANGUAGE BangPatterns, CPP #-}
1+{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables #-}
2
2
3module Network.Transport.Pipes
3module Network.Transport.Pipes
4 ( mkTransport
4 ( mkTransport
@@ -6,7 +6,7 @@ module Network.Transport.Pipes
6
6
7import Control.Monad (when)
7import Control.Monad (when)
8import Control.Concurrent.MVar
8import Control.Concurrent.MVar
9−import Control.Concurrent (threadDelay)
9+import Control.Concurrent (threadDelay, forkOS)
10import Data.IntMap (IntMap)
10import Data.IntMap (IntMap)
11-- import Data.ByteString.Char8 (ByteString)
11-- import Data.ByteString.Char8 (ByteString)
12import Data.Word
12import Data.Word
@@ -19,12 +19,14 @@ import Data.Serialize (encode,decode)
19import Data.List (foldl')
19import Data.List (foldl')
20import Network.Transport
20import Network.Transport
21import System.Random (randomIO)
21import System.Random (randomIO)
22−import System.IO (IOMode(ReadMode,AppendMode), openFile, hClose)
22+import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
23+ openFile, hClose, hPutStrLn, hPutStr, stderr, stdout)
23import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
24import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
24
25
25-- Option 1: low level unix routines:
26-- Option 1: low level unix routines:
26-- Option 2: ByteString-provided IO routines
27-- Option 2: ByteString-provided IO routines
28+#define UNIXIO
27−#if UNIXIO
29+#ifdef UNIXIO
28-- Added in unix-2.5.1.0. Let's not depend on this yet:
30-- Added in unix-2.5.1.0. Let's not depend on this yet:
29-- import qualified System.Posix.IO.ByteString as PIO
31-- import qualified System.Posix.IO.ByteString as PIO
30import qualified System.Posix.IO as PIO
32import qualified System.Posix.IO as PIO
@@ -47,16 +49,14 @@ mkTransport = do
47 lock <- newMVar ()
49 lock <- newMVar ()
48 let filename = "/tmp/pipe_"++show uid
50 let filename = "/tmp/pipe_"++show uid
49 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
51 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
50−#if UNIXIO
52+
51− pipe <- PIO.openFd filename PIO.ReadOnly Nothing PIO.defaultFileFlags
53+ dbgprint1$ " Created pipe at location: "++ filename
52−#else
53− pipe <- openFile filename ReadMode
54−#endif
55
54
56 return Transport
55 return Transport
57 { newConnectionWith = \ _ -> do
56 { newConnectionWith = \ _ -> do
57+ dbgprint1$ " Creating new connection"
58 return (mkSourceAddr filename,
58 return (mkSourceAddr filename,
59− mkTargetEnd pipe lock)
59+ mkTargetEnd filename lock)
60 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
60 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
61 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
61 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
62 }
62 }
@@ -70,52 +70,85 @@ mkTransport = do
70 mkSourceEnd :: String -> IO SourceEnd
70 mkSourceEnd :: String -> IO SourceEnd
71 mkSourceEnd filename = do
71 mkSourceEnd filename = do
72-- fd <- openFile filename AppendMode
72-- fd <- openFile filename AppendMode
73+-- mv <- onOSThread$ openFile filename WriteMode
73 return $
74 return $
74 -- Write to the named pipe. If the message is less than
75 -- Write to the named pipe. If the message is less than
75 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
76 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
76 -- we have to do something more sophisticated.
77 -- we have to do something more sophisticated.
77 SourceEnd
78 SourceEnd
78 { send = \bss -> do
79 { send = \bss -> do
79− putStrLn$ "SENDING ... "++ show bss
80+ dbgprint1$ "SENDING ... "++ show bss
80
81
81 -- This may happen on multiple processes/threads:
82 -- This may happen on multiple processes/threads:
82− let msgsize = foldl' (\n s -> n + BS.length s) 0 bss
83+ let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss
83 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
84 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
84 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
85 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
85 -- Otherwise it's just a simple write:
86 -- Otherwise it's just a simple write:
86 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
87 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
88+
89+-- fd <- readMVar mv
90+ dbgprint1$ " (sending... got file descriptor) "
87− fd <- openFile filename AppendMode
91+-- fd <- openFile filename AppendMode
92+ fd <- openFile filename WriteMode
93+
94+-- TODO: Consider nonblocking opening of file to make things simpler.
95+
88− BS.hPut fd (BS.concat (encode msgsize : bss))
96+ let finalmsg = BS.concat (encode msgsize : bss)
97+ dbgprint1$ " Final send msg: " ++ show finalmsg
98+ BS.hPut fd finalmsg
89 hClose fd -- TEMP -- opening and closing on each send!
99 hClose fd -- TEMP -- opening and closing on each send!
90 }
100 }
91
101
92-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
102-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
93− mkTargetEnd fd lock = TargetEnd
103+ mkTargetEnd filename lock = TargetEnd
94 { receive = do
104 { receive = do
105+ dbgprint2$ "Begin receive action..."
95 -- This should only happen on a single process. But it may
106 -- This should only happen on a single process. But it may
96 -- happen on multiple threads so we grab a lock.
107 -- happen on multiple threads so we grab a lock.
97 takeMVar lock
108 takeMVar lock
98#ifdef UNIXIO
109#ifdef UNIXIO
99− (bytes,cnt) <- PIO.fdRead fd sizeof_header
110+-- fd <- PIO.openFd filename PIO.ReadOnly Nothing PIO.defaultFileFlags
111+ fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
112+ let oneread n = do (s,cnt) <- PIO.fdRead fd (fromIntegral n)
113+ return (BS.pack s, fromIntegral cnt)
100#else
114#else
115+ fd <- openFile filename ReadMode
116+ let oneread n = do bs <- BS.hGet fd n;
117+ return (bs, BS.length bs)
118+#endif
101− putStrLn$ " Attempt read header..."
119+ dbgprint2$ " Attempt read header..."
102
120
103− let rdloop = do
121+ let spinread :: Int -> IO BS.ByteString
104− putStr "."
122+ spinread desired = do
105− hdr <- BS.hGet fd sizeof_header
123+-- hPutStr stderr "."
124+ (bytes,len) <- oneread desired
106− case BS.length hdr of
125+ case len of
107− n | n == sizeof_header -> return hdr
126+ n | n == desired -> return bytes
108 0 -> do threadDelay (10*1000)
127 0 -> do threadDelay (10*1000)
109− rdloop
128+ spinread desired
110− l -> error$ "Inclomplete read of msg header, only "++ show l ++ " bytes"
129+ l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
111− hdr <- rdloop
130+ show desired ++" bytes) got "++ show l ++ " bytes"
131+
132+ hdr <- spinread sizeof_header
112− putStrLn$ " Got header "++ show hdr ++ " attempt read payload"
133+ dbgprint2$ " Got header "++ show hdr ++ " attempt read payload"
113− payload <- case decode hdr of
134+ payload <- case decode hdr of
114− Left err -> error$ "ERROR: "++ err
135+ Left err -> error$ "ERROR: "++ err
115− Right size -> BS.hGet fd size
136+ Right size -> spinread (fromIntegral (size::Word32))
116− putStrLn$ " Got payload "++ show payload
137+ dbgprint2$ " Got payload "++ show payload
117−#endif
138+
118 putMVar lock ()
139 putMVar lock ()
119 return [payload]
140 return [payload]
120 }
141 }
121
142
143+-- dbgprint2 = hPutStrLn stdout
144+-- dbgprint1 = hPutStrLn stderr
145+dbgprint1 _ = return ()
146+dbgprint2 _ = return ()
147+
148+
149+-- Execute an action on its own OS thread. Return an MVar to synchronize on.
150+onOSThread :: IO a -> IO (MVar a)
151+onOSThread action = do
152+ mv <- newEmptyMVar
153+ forkOS (action >>= putMVar mv )
154+ return mv