@@ -1,4 +1,4 @@
1−{-# LANGUAGE BangPatterns, CPP #-}
1+{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables #-}
22
33module Network.Transport.Pipes
44 ( mkTransport
@@ -6,7 +6,7 @@ module Network.Transport.Pipes
66
77import Control.Monad (when)
88import Control.Concurrent.MVar
9−import Control.Concurrent (threadDelay)
9+import Control.Concurrent (threadDelay, forkOS)
1010import Data.IntMap (IntMap)
1111-- import Data.ByteString.Char8 (ByteString)
1212import Data.Word
@@ -19,12 +19,14 @@ import Data.Serialize (encode,decode)
1919import Data.List (foldl')
2020import Network.Transport
2121import 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)
2324import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
2425
2526-- Option 1: low level unix routines:
2627-- Option 2: ByteString-provided IO routines
27−#if UNIXIO
28+#define UNIXIO
29+#ifdef UNIXIO
2830-- Added in unix-2.5.1.0. Let's not depend on this yet:
2931-- import qualified System.Posix.IO.ByteString as PIO
3032import qualified System.Posix.IO as PIO
@@ -47,16 +49,14 @@ mkTransport = do
4749 lock <- newMVar ()
4850 let filename = "/tmp/pipe_"++show uid
4951 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
50−#if UNIXIO
51− pipe <- PIO.openFd filename PIO.ReadOnly Nothing PIO.defaultFileFlags
52−#else
53− pipe <- openFile filename ReadMode
54−#endif
52+
53+ dbgprint1$ " Created pipe at location: "++ filename
5554
5655 return Transport
5756 { newConnectionWith = \ _ -> do
57+ dbgprint1$ " Creating new connection"
5858 return (mkSourceAddr filename,
59− mkTargetEnd pipe lock)
59+ mkTargetEnd filename lock)
6060 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
6161 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
6262 }
@@ -70,52 +70,85 @@ mkTransport = do
7070 mkSourceEnd :: String -> IO SourceEnd
7171 mkSourceEnd filename = do
7272-- fd <- openFile filename AppendMode
73+-- mv <- onOSThread$ openFile filename WriteMode
7374 return $
7475 -- Write to the named pipe. If the message is less than
7576 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
7677 -- we have to do something more sophisticated.
7778 SourceEnd
7879 { send = \bss -> do
79− putStrLn$ "SENDING ... "++ show bss
80+ dbgprint1$ "SENDING ... "++ show bss
8081
8182 -- 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
8384 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
8485 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
8586 -- Otherwise it's just a simple write:
8687 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
87− fd <- openFile filename AppendMode
88− BS.hPut fd (BS.concat (encode msgsize : bss))
88+
89+-- fd <- readMVar mv
90+ dbgprint1$ " (sending... got file descriptor) "
91+-- fd <- openFile filename AppendMode
92+ fd <- openFile filename WriteMode
93+
94+-- TODO: Consider nonblocking opening of file to make things simpler.
95+
96+ let finalmsg = BS.concat (encode msgsize : bss)
97+ dbgprint1$ " Final send msg: " ++ show finalmsg
98+ BS.hPut fd finalmsg
8999 hClose fd -- TEMP -- opening and closing on each send!
90100 }
91101
92102-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
93− mkTargetEnd fd lock = TargetEnd
103+ mkTargetEnd filename lock = TargetEnd
94104 { receive = do
105+ dbgprint2$ "Begin receive action..."
95106 -- This should only happen on a single process. But it may
96107 -- happen on multiple threads so we grab a lock.
97108 takeMVar lock
98109#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)
100114#else
101− putStrLn$ " Attempt read header..."
115+ fd <- openFile filename ReadMode
116+ let oneread n = do bs <- BS.hGet fd n;
117+ return (bs, BS.length bs)
118+#endif
119+ dbgprint2$ " Attempt read header..."
102120
103− let rdloop = do
104− putStr "."
105− hdr <- BS.hGet fd sizeof_header
106− case BS.length hdr of
107− n | n == sizeof_header -> return hdr
121+ let spinread :: Int -> IO BS.ByteString
122+ spinread desired = do
123+-- hPutStr stderr "."
124+ (bytes,len) <- oneread desired
125+ case len of
126+ n | n == desired -> return bytes
108127 0 -> do threadDelay (10*1000)
109− rdloop
110− l -> error$ "Inclomplete read of msg header, only "++ show l ++ " bytes"
111− hdr <- rdloop
112− putStrLn$ " Got header "++ show hdr ++ " attempt read payload"
113− payload <- case decode hdr of
114− Left err -> error$ "ERROR: "++ err
115− Right size -> BS.hGet fd size
116− putStrLn$ " Got payload "++ show payload
117−#endif
128+ spinread desired
129+ l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
130+ show desired ++" bytes) got "++ show l ++ " bytes"
131+
132+ hdr <- spinread sizeof_header
133+ dbgprint2$ " Got header "++ show hdr ++ " attempt read payload"
134+ payload <- case decode hdr of
135+ Left err -> error$ "ERROR: "++ err
136+ Right size -> spinread (fromIntegral (size::Word32))
137+ dbgprint2$ " Got payload "++ show payload
138+
118139 putMVar lock ()
119140 return [payload]
120141 }
121142
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