git.iko.soy/distributed-process
on master help sign in
Got to the bottom of the regression in the Pipes transport.
d72c0f5Ryan Newton
examples/DemoTransport.hs+20−10
compare with master
unifiedsplit
@@ -133,10 +133,18 @@ logServer name targetEnd = forever $ do
133 trace (name ++ " rcvd: " ++ show x) $ return ()
133 trace (name ++ " rcvd: " ++ show x) $ return ()
134
134
135
135
136+
137+-- do cnt <- readIORef cntr
138+-- writeIORef cntr (cnt+1)
139+-- mkTransport (TCPConfig undefined "127.0.0.1" (show (8080 + offset + cnt)))
140+
141+--mkTCP = do cntr <- newIORef 0
142+-- mkTCPOff 8080
143+
136runWAllTranports :: (IO Transport -> IO ()) -> Int -> IO ()
144runWAllTranports :: (IO Transport -> IO ()) -> Int -> IO ()
137runWAllTranports demo offset = do
145runWAllTranports demo offset = do
138 putStrLn "------------------------------------------------------------"
146 putStrLn "------------------------------------------------------------"
139−{-
147+
140 putStrLn " MVAR transport:"
148 putStrLn " MVAR transport:"
141 demo Network.Transport.MVar.mkTransport
149 demo Network.Transport.MVar.mkTransport
142
150
@@ -145,28 +153,30 @@ runWAllTranports demo offset = do
145 demo$ do cnt <- readIORef cntr
153 demo$ do cnt <- readIORef cntr
146 writeIORef cntr (cnt+1)
154 writeIORef cntr (cnt+1)
147 mkTransport (TCPConfig undefined "127.0.0.1" (show (8080 + offset + cnt)))
155 mkTransport (TCPConfig undefined "127.0.0.1" (show (8080 + offset + cnt)))
148−-}
156+
149 putStrLn "\n PIPES transport:"
157 putStrLn "\n PIPES transport:"
150 demo Network.Transport.Pipes.mkTransport
158 demo Network.Transport.Pipes.mkTransport
151 putStrLn "\n"
159 putStrLn "\n"
152
160
153main = do
161main = do
162+
154 putStrLn "Demo0:"
163 putStrLn "Demo0:"
155 runWAllTranports demo0 0
164 runWAllTranports demo0 0
156−{-
165+
157 putStrLn "Demo1:"
166 putStrLn "Demo1:"
158 runWAllTranports demo1 10
167 runWAllTranports demo1 10
159
168
160− putStrLn "Demo2:"
169+-- putStrLn "Demo2:"
161− runWAllTranports demo2 20
170+-- runWAllTranports demo2 20
162
171
163− putStrLn "Demo3:"
172+-- putStrLn "Demo3:"
164− runWAllTranports demo3 30
173+-- runWAllTranports demo3 30
174+
175+-- putStrLn "Demo4:"
176+-- runWAllTranports demo4 40
165
177
166− putStrLn "Demo4:"
167− runWAllTranports demo4 40
168−-}
169 threadDelay (500 * 1000)
178 threadDelay (500 * 1000)
179+
170 putStrLn "Done with all demos!"
180 putStrLn "Done with all demos!"
171
181
172-- trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
182-- trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
network-transport/src/Network/Transport/Pipes.hs+35−9
@@ -5,6 +5,7 @@ module Network.Transport.Pipes
5 ) where
5 ) where
6
6
7import Control.Monad (when, unless)
7import Control.Monad (when, unless)
8+import Control.Exception (evaluate)
8import Control.Concurrent.MVar
9import Control.Concurrent.MVar
9import Control.Concurrent (threadDelay, forkOS)
10import Control.Concurrent (threadDelay, forkOS)
10import Data.IntMap (IntMap)
11import Data.IntMap (IntMap)
@@ -16,8 +17,14 @@ import qualified Data.IntMap as IntMap
16import qualified Data.ByteString.Char8 as BSS
17import qualified Data.ByteString.Char8 as BSS
17import qualified Data.ByteString.Lazy.Char8 as BS
18import qualified Data.ByteString.Lazy.Char8 as BS
18
19
19−import Data.Binary (encode,decode)
20+-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:
21+#define CEREAL
22+#ifdef CEREAL
20−-- import Data.Serialize (encode,decode)
23+import Data.Serialize (encode,decode) -- Uses strict BS
24+#else
25+import Data.Binary (encode,decode) -- Uses lazy BS
26+#endif
27+
21import Data.List (foldl')
28import Data.List (foldl')
22import Network.Transport
29import Network.Transport
23import System.Random (randomIO)
30import System.Random (randomIO)
@@ -27,6 +34,7 @@ import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, owner
27import System.Posix.Types (Fd)
34import System.Posix.Types (Fd)
28
35
29-- define DEBUG
36-- define DEBUG
37+-- define USE_UNIX_BYTESTRING
30
38
31#ifdef USE_UNIX_BYTESTRING
39#ifdef USE_UNIX_BYTESTRING
32import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
40import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
@@ -79,11 +87,19 @@ mkTransport = do
79 -- Initiate but do not block on file opening:
87 -- Initiate but do not block on file opening:
80 -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.
88 -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.
81 --
89 --
90+
91+ -- All THREE of the below options were observed to work on a simple demo:
92+
93+ -- OPTION (1)
82− -- We may protect from blocking other threads by running on a separate (OS) thread:
94+ -- Here we protect from blocking other threads by running on a separate (OS) thread:
83-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
95-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
96+-- fd <- takeMVar mv
97+
98+ -- OPTION (2)
84-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
99-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
100+
101+ -- OPTION (3)
85 fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags
102 fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags
86−-- fd <- takeMVar mv
87 dbgprint1$ "GOT WRITING END OPEN ... "
103 dbgprint1$ "GOT WRITING END OPEN ... "
88
104
89 return $
105 return $
@@ -92,7 +108,7 @@ mkTransport = do
92 -- we have to do something more sophisticated.
108 -- we have to do something more sophisticated.
93 SourceEnd
109 SourceEnd
94 { send = \bss -> do
110 { send = \bss -> do
95− dbgprint1$ "SENDING ... "++ show bss
111+ dbgprint1$ "Sending.. "++ show bss
96
112
97 -- This may happen on multiple processes/threads:
113 -- This may happen on multiple processes/threads:
98 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss
114 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss
@@ -100,7 +116,8 @@ mkTransport = do
100 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
116 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
101 -- Otherwise it's just a simple write:
117 -- Otherwise it's just a simple write:
102 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
118 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
103− let finalmsg = BS.concat (encode msgsize : bss)
119+-- let finalmsg = BS.concat (encode msgsize : bss)
120+ let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)
104 dbgprint1$ " Final send msg: " ++ show finalmsg
121 dbgprint1$ " Final send msg: " ++ show finalmsg
105
122
106 -- OPTION 2: Speculative file opening, plus this synchronnization:
123 -- OPTION 2: Speculative file opening, plus this synchronnization:
@@ -132,7 +149,8 @@ mkTransport = do
132 let spinread :: Int64 -> IO BS.ByteString
149 let spinread :: Int64 -> IO BS.ByteString
133 spinread desired = do
150 spinread desired = do
134#ifdef DEBUG
151#ifdef DEBUG
135−-- hPutStr stderr "."
152+ hPutStr stderr "."
153+-- dbgprint2$ " SPINREAD "++ show desired
136 BSS.hPutStr stdout (BSS.pack$ " "++show desired)
154 BSS.hPutStr stdout (BSS.pack$ " "++show desired)
137#endif
155#endif
138 (bytes,len) <- oneread desired
156 (bytes,len) <- oneread desired
@@ -144,8 +162,16 @@ mkTransport = do
144 show desired ++" bytes) got "++ show l ++ " bytes"
162 show desired ++" bytes) got "++ show l ++ " bytes"
145
163
146 hdr <- spinread sizeof_header
164 hdr <- spinread sizeof_header
147− dbgprint2$ " Got header "++ show hdr ++ " attempt read payload"
165+ dbgprint2$ " Got header "++ show hdr ++ ", next attempt to read payload:"
166+#ifdef CEREAL
167+ let decoded = decode (BSS.concat$ BS.toChunks hdr)
168+-- dbgprint2$ " DECODING HDR, bytes "++ show (BS.length hdr) ++ ": "++show hdr
169+-- evaluate decoded
170+-- dbgprint2$ " DONE DECODING HDR"
171+#else
172+ let decoded = decode hdr
173+#endif
148− payload <- case decode hdr of
174+ payload <- case decoded of
149 Left err -> error$ "ERROR: "++ err
175 Left err -> error$ "ERROR: "++ err
150 Right size -> spinread (fromIntegral (size::Word32))
176 Right size -> spinread (fromIntegral (size::Word32))
151 dbgprint2$ " Got payload "++ show payload
177 dbgprint2$ " Got payload "++ show payload