git.iko.soy/distributed-process
on master help sign in
This is still covered in debugging cruft, but it works without opening
1611171Ryan Newton
examples/DemoTransport2.hs+7−2
compare with master
unifiedsplit
@@ -1,3 +1,5 @@
1+{-# LANGUAGE CPP #-}
2+
1-- module DemoTransport where
3-- module DemoTransport where
2module Main where
4module Main where
3
5
@@ -38,9 +40,12 @@ demo0 mkTrans = do
38
40
39 (sourceAddr, targetEnd) <- newConnection trans
41 (sourceAddr, targetEnd) <- newConnection trans
40
42
41−-- forkWithExceptions forkIO "receiver thread" $ logServer "logServer" targetEnd
43+#ifdef EXNTREE
44+ -- [2012.02.19] I'm seeing indefinite MVar errors cropping up here:
42−-- forkWithExceptions forkOS "receiver thread" $ logServer "logServer" targetEnd
45+ forkWithExceptions forkOS "receiver thread" $ logServer "logServer" targetEnd
46+#else
43 forkIO $ logServer "logServer" targetEnd
47 forkIO $ logServer "logServer" targetEnd
48+#endif
44 threadDelay 100000
49 threadDelay 100000
45
50
46 sourceEnd <- connect sourceAddr
51 sourceEnd <- connect sourceAddr
network-transport/src/Network/Transport/Pipes.hs+46−9
@@ -4,12 +4,13 @@ module Network.Transport.Pipes
4 ( mkTransport
4 ( mkTransport
5 ) where
5 ) where
6
6
7−import Control.Monad (when)
7+import Control.Monad (when, unless)
8import Control.Concurrent.MVar
8import Control.Concurrent.MVar
9import Control.Concurrent (threadDelay, forkOS)
9import 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
13+import Data.IORef
13import qualified Data.IntMap as IntMap
14import qualified Data.IntMap as IntMap
14import qualified Data.ByteString.Char8 as BS
15import qualified Data.ByteString.Char8 as BS
15-- import qualified Data.ByteString.Lazy.Char8 as BS
16-- import qualified Data.ByteString.Lazy.Char8 as BS
@@ -69,8 +70,16 @@ mkTransport = do
69
70
70 mkSourceEnd :: String -> IO SourceEnd
71 mkSourceEnd :: String -> IO SourceEnd
71 mkSourceEnd filename = do
72 mkSourceEnd filename = do
73+# ifndef KEEP_CLOSED
72-- fd <- openFile filename AppendMode
74-- fd <- openFile filename AppendMode
75+-- _ <- openFile filename WriteMode
73-- mv <- onOSThread$ openFile filename WriteMode
76-- mv <- onOSThread$ openFile filename WriteMode
77+ mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
78+
79+ -- [2012.02.19] Still getting indefinite MVar block errors even
80+ -- using IORefs here:
81+-- fdref <- newIORef Nothing
82+# endif
74 return $
83 return $
75 -- Write to the named pipe. If the message is less than
84 -- Write to the named pipe. If the message is less than
76 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
85 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
@@ -85,20 +94,43 @@ mkTransport = do
85 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
94 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
86 -- Otherwise it's just a simple write:
95 -- Otherwise it's just a simple write:
87 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
96 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
88−
97+
89−-- fd <- readMVar mv
98+# ifdef KEEP_CLOSED
90− dbgprint1$ " (sending... got file descriptor) "
91-- fd <- openFile filename AppendMode
99-- fd <- openFile filename AppendMode
92 fd <- openFile filename WriteMode
100 fd <- openFile filename WriteMode
101+# else
102+ -- OPTION 1: Lazy opening:
103+ -- r <- readIORef fdref
104+ -- fd <- case r of
105+ -- Nothing -> do
106+ -- dbgprint1$ " (No FD for sending yet... opening) "
107+ -- fd <- openFile filename AppendMode
108+ -- writeIORef fdref (Just fd)
109+ -- return fd
110+ -- Just fd -> return fd
111+
112+ -- OPTION 2: Speculative file opening, plus this synchronnization:
113+ fd <- readMVar mv
114+ dbgprint1$ " (sending... got file descriptor) "
115+# endif
93
116
94-- TODO: Consider nonblocking opening of file to make things simpler.
117-- TODO: Consider nonblocking opening of file to make things simpler.
95
118
96 let finalmsg = BS.concat (encode msgsize : bss)
119 let finalmsg = BS.concat (encode msgsize : bss)
97 dbgprint1$ " Final send msg: " ++ show finalmsg
120 dbgprint1$ " Final send msg: " ++ show finalmsg
121+ ----------------------------------------
98− BS.hPut fd finalmsg
122+ -- BS.hPut fd finalmsg -- The actual send!
123+ cnt <- PIO.fdWrite fd (BS.unpack finalmsg) -- inefficient!
124+ unless (fromIntegral cnt == BS.length finalmsg) $
125+ error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
126+ ----------------------------------------
127+# ifdef KEEP_CLOSED
99 hClose fd -- TEMP -- opening and closing on each send!
128 hClose fd -- TEMP -- opening and closing on each send!
129+# endif
130+ return ()
100 }
131 }
101
132
133+
102-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
134-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
103 mkTargetEnd filename lock = TargetEnd
135 mkTargetEnd filename lock = TargetEnd
104 { receive = do
136 { receive = do
@@ -108,7 +140,8 @@ mkTransport = do
108 takeMVar lock
140 takeMVar lock
109#ifdef UNIXIO
141#ifdef UNIXIO
110-- fd <- PIO.openFd filename PIO.ReadOnly Nothing PIO.defaultFileFlags
142-- fd <- PIO.openFd filename PIO.ReadOnly Nothing PIO.defaultFileFlags
111− fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
143+ mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
144+ fd <- takeMVar mv
112 let oneread n = do (s,cnt) <- PIO.fdRead fd (fromIntegral n)
145 let oneread n = do (s,cnt) <- PIO.fdRead fd (fromIntegral n)
113 return (BS.pack s, fromIntegral cnt)
146 return (BS.pack s, fromIntegral cnt)
114#else
147#else
@@ -117,7 +150,7 @@ mkTransport = do
117 return (bs, BS.length bs)
150 return (bs, BS.length bs)
118#endif
151#endif
119 dbgprint2$ " Attempt read header..."
152 dbgprint2$ " Attempt read header..."
120−
153+
121 let spinread :: Int -> IO BS.ByteString
154 let spinread :: Int -> IO BS.ByteString
122 spinread desired = do
155 spinread desired = do
123-- hPutStr stderr "."
156-- hPutStr stderr "."
@@ -140,10 +173,13 @@ mkTransport = do
140 return [payload]
173 return [payload]
141 }
174 }
142
175
176+#ifdef DEBUG
143−-- dbgprint2 = hPutStrLn stdout
177+dbgprint2 = hPutStrLn stdout
144−-- dbgprint1 = hPutStrLn stderr
178+dbgprint1 = hPutStrLn stderr
179+#else
145dbgprint1 _ = return ()
180dbgprint1 _ = return ()
146dbgprint2 _ = return ()
181dbgprint2 _ = return ()
182+#endif
147
183
148
184
149-- Execute an action on its own OS thread. Return an MVar to synchronize on.
185-- Execute an action on its own OS thread. Return an MVar to synchronize on.
@@ -152,3 +188,4 @@ onOSThread action = do
152 mv <- newEmptyMVar
188 mv <- newEmptyMVar
153 forkOS (action >>= putMVar mv )
189 forkOS (action >>= putMVar mv )
154 return mv
190 return mv
191+-- [2012.02.19] This didn't seem to help.