@@ -0,0 +1,121 @@
1+{-# LANGUAGE BangPatterns, CPP #-}
2+
3+module Network.Transport.Pipes
4+ ( mkTransport
5+ ) where
6+
7+import Control.Monad (when)
8+import Control.Concurrent.MVar
9+import Control.Concurrent (threadDelay)
10+import Data.IntMap (IntMap)
11+-- import Data.ByteString.Char8 (ByteString)
12+import Data.Word
13+import qualified Data.IntMap as IntMap
14+import qualified Data.ByteString.Char8 as BS
15+-- import qualified Data.ByteString.Lazy.Char8 as BS
16+
17+-- import Data.Binary (decode)
18+import Data.Serialize (encode,decode)
19+import Data.List (foldl')
20+import Network.Transport
21+import System.Random (randomIO)
22+import System.IO (IOMode(ReadMode,AppendMode), openFile, hClose)
23+import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
24+
25+-- Option 1: low level unix routines:
26+-- Option 2: ByteString-provided IO routines
27+#if UNIXIO
28+-- Added in unix-2.5.1.0. Let's not depend on this yet:
29+-- import qualified System.Posix.IO.ByteString as PIO
30+import qualified System.Posix.IO as PIO
31+import System.Posix.Types (Fd)
32+#else
33+
34+#endif
35+
36+
37+-- An address is just a filename -- the named pipe.
38+data Addr = Addr String
39+ deriving (Show,Read)
40+
41+-- The msg header consists of just a length field represented as a Word32
42+sizeof_header = 4
43+
44+mkTransport :: IO Transport
45+mkTransport = do
46+ uid <- randomIO :: IO Word64
47+ lock <- newMVar ()
48+ let filename = "/tmp/pipe_"++show uid
49+ 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
55+
56+ return Transport
57+ { newConnectionWith = \ _ -> do
58+ return (mkSourceAddr filename,
59+ mkTargetEnd pipe lock)
60+ , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
61+ , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
62+ }
63+ where
64+ mkSourceAddr :: String -> SourceAddr
65+ mkSourceAddr filename = SourceAddr
66+ { connectWith = \_ -> mkSourceEnd filename
67+ , serialize = BS.pack filename
68+ }
69+
70+ mkSourceEnd :: String -> IO SourceEnd
71+ mkSourceEnd filename = do
72+-- fd <- openFile filename AppendMode
73+ return $
74+ -- Write to the named pipe. If the message is less than
75+ -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
76+ -- we have to do something more sophisticated.
77+ SourceEnd
78+ { send = \bss -> do
79+ putStrLn$ "SENDING ... "++ show bss
80+
81+ -- This may happen on multiple processes/threads:
82+ let msgsize = foldl' (\n s -> n + BS.length s) 0 bss
83+ 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+ -- Otherwise it's just a simple write:
86+ -- 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))
89+ hClose fd -- TEMP -- opening and closing on each send!
90+ }
91+
92+-- mkTargetEnd :: Fd -> MVar () -> TargetEnd
93+ mkTargetEnd fd lock = TargetEnd
94+ { receive = do
95+ -- This should only happen on a single process. But it may
96+ -- happen on multiple threads so we grab a lock.
97+ takeMVar lock
98+#ifdef UNIXIO
99+ (bytes,cnt) <- PIO.fdRead fd sizeof_header
100+#else
101+ putStrLn$ " Attempt read header..."
102+
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
108+ 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
118+ putMVar lock ()
119+ return [payload]
120+ }
121+