d72c0f5 · distributed-process

git.iko.soy/distributed-process

Got to the bottom of the regression in the Pipes transport.

The catalyst was switching between cereal and binary. I have no idea (still) why this is. This checkin switches back to cereal (and inefficiently converts lazy/strict bytestrings). Demo 0 and 1 work for all transports. The rest have remaining problems for Pipes.

commitd72c0f5bb6528e85c31c88c12d28877e12def445 authorRyan Newton <rrnewton@gmail.com> dateMon 20 Feb 2012, 16:51 (15 years ago) parent8c065e4

2 files changed+55−19

compare with master

unifiedsplit

examples/DemoTransport.hs+20−10 network-transport/src/Network/Transport/Pipes.hs+35−9

examples/DemoTransport.hs+20−10

@@ -133,10 +133,18 @@ logServer name targetEnd = forever $ do

133133 trace (name ++ " rcvd: " ++ show x) $ return ()

134134

135135

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+

136144runWAllTranports :: (IO Transport -> IO ()) -> Int -> IO ()

137145runWAllTranports demo offset = do

138146 putStrLn "------------------------------------------------------------"

139−{-

147+

140148 putStrLn " MVAR transport:"

141149 demo Network.Transport.MVar.mkTransport

142150

@@ -145,28 +153,30 @@ runWAllTranports demo offset = do

145153 demo$ do cnt <- readIORef cntr

146154 writeIORef cntr (cnt+1)

147155 mkTransport (TCPConfig undefined "127.0.0.1" (show (8080 + offset + cnt)))

148−-}

156+

149157 putStrLn "\n PIPES transport:"

150158 demo Network.Transport.Pipes.mkTransport

151159 putStrLn "\n"

152160

153161main = do

162+

154163 putStrLn "Demo0:"

155164 runWAllTranports demo0 0

156−{-

165+

157166 putStrLn "Demo1:"

158167 runWAllTranports demo1 10

159168

160− putStrLn "Demo2:"

161− runWAllTranports demo2 20

169+-- putStrLn "Demo2:"

170+-- runWAllTranports demo2 20

162171

163− putStrLn "Demo3:"

164− runWAllTranports demo3 30

172+-- putStrLn "Demo3:"

173+-- runWAllTranports demo3 30

174+

175+-- putStrLn "Demo4:"

176+-- runWAllTranports demo4 40

165177

166− putStrLn "Demo4:"

167− runWAllTranports demo4 40

168−-}

169178 threadDelay (500 * 1000)

179+

170180 putStrLn "Done with all demos!"

171181

172182-- 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

55 ) where

77import Control.Monad (when, unless)

8+import Control.Exception (evaluate)

89import Control.Concurrent.MVar

910import Control.Concurrent (threadDelay, forkOS)

1011import Data.IntMap (IntMap)

@@ -16,8 +17,14 @@ import qualified Data.IntMap as IntMap

1617import qualified Data.ByteString.Char8 as BSS

1718import qualified Data.ByteString.Lazy.Char8 as BS

1819

19−import Data.Binary (encode,decode)

20−-- import Data.Serialize (encode,decode)

20+-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:

21+#define CEREAL

22+#ifdef CEREAL

23+import Data.Serialize (encode,decode) -- Uses strict BS

24+#else

25+import Data.Binary (encode,decode) -- Uses lazy BS

26+#endif

27+

2128import Data.List (foldl')

2229import Network.Transport

2330import System.Random (randomIO)

@@ -27,6 +34,7 @@ import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, owner

2734import System.Posix.Types (Fd)

2835

2936-- define DEBUG

37+-- define USE_UNIX_BYTESTRING

3038

3139#ifdef USE_UNIX_BYTESTRING

3240import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO

@@ -79,11 +87,19 @@ mkTransport = do

7987 -- Initiate but do not block on file opening:

8088 -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.

8189 --

82− -- We may protect from blocking other threads by running on a separate (OS) thread:

90+

91+ -- All THREE of the below options were observed to work on a simple demo:

92+

93+ -- OPTION (1)

94+ -- Here we protect from blocking other threads by running on a separate (OS) thread:

8395-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags

96+-- fd <- takeMVar mv

97+

98+ -- OPTION (2)

8499-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags

100+

101+ -- OPTION (3)

85102 fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags

86−-- fd <- takeMVar mv

87103 dbgprint1$ "GOT WRITING END OPEN ... "

88104

89105 return $

@@ -92,7 +108,7 @@ mkTransport = do

92108 -- we have to do something more sophisticated.

93109 SourceEnd

94110 { send = \bss -> do

95− dbgprint1$ "SENDING ... "++ show bss

111+ dbgprint1$ "Sending.. "++ show bss

96112

97113 -- This may happen on multiple processes/threads:

98114 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss

@@ -100,7 +116,8 @@ mkTransport = do

100116 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."

101117 -- Otherwise it's just a simple write:

102118 -- 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)

104121 dbgprint1$ " Final send msg: " ++ show finalmsg

105122

106123 -- OPTION 2: Speculative file opening, plus this synchronnization:

@@ -132,7 +149,8 @@ mkTransport = do

132149 let spinread :: Int64 -> IO BS.ByteString

133150 spinread desired = do

134151#ifdef DEBUG

135−-- hPutStr stderr "."

152+ hPutStr stderr "."

153+-- dbgprint2$ " SPINREAD "++ show desired

136154 BSS.hPutStr stdout (BSS.pack$ " "++show desired)

137155#endif

138156 (bytes,len) <- oneread desired

@@ -144,8 +162,16 @@ mkTransport = do

144162 show desired ++" bytes) got "++ show l ++ " bytes"

145163

146164 hdr <- spinread sizeof_header

147− dbgprint2$ " Got header "++ show hdr ++ " attempt read payload"

148− payload <- case decode hdr of

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

174+ payload <- case decoded of

149175 Left err -> error$ "ERROR: "++ err

150176 Right size -> spinread (fromIntegral (size::Word32))

151177 dbgprint2$ " Got payload "++ show payload