@@ -0,0 +1,235 @@
1+{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables, PackageImports #-}
2+
3+----------------------------------------------------------------------------------------------------
4+
5+-- WARNING -- This transport is not yet completed. TODO:
6+
7+-- * Add support for messages greater than 4096 bytes.
8+-- * debug ODD problem with CEREAL below
9+-- * switch to unix-bytestring after that package is updated for 7.4.1
10+
11+----------------------------------------------------------------------------------------------------
12+
13+module Network.Transport.Pipes
14+ ( mkTransport
15+ ) where
16+
17+import Control.Monad (when, unless)
18+import Control.Exception (evaluate, throw, handle, fromException,
19+ SomeException, IOException)
20+import Control.Concurrent.MVar
21+import Control.Concurrent (threadDelay, forkOS)
22+import Data.IntMap (IntMap)
23+-- import Data.ByteString.Char8 (ByteString)
24+import Data.Word
25+import Data.Int
26+import Data.IORef
27+import qualified Data.IntMap as IntMap
28+import qualified Data.ByteString.Char8 as BSS
29+import qualified Data.ByteString.Lazy.Char8 as BS
30+
31+-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:
32+#define CEREAL
33+#ifdef CEREAL
34+import Data.Serialize (encode,decode) -- Uses strict BS
35+#else
36+import Data.Binary (encode,decode) -- Uses lazy BS
37+#endif
38+
39+import Data.List (foldl')
40+import Network.Transport
41+import System.Random (randomIO)
42+import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
43+ openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
44+import System.IO.Error (ioeGetHandle)
45+import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
46+import System.Posix.Types (Fd, ByteCount)
47+import System.Directory (doesFileExist, removeFile)
48+import System.Mem (performGC)
49+
50+#define USE_UNIX_BYTESTRING
51+#ifdef USE_UNIX_BYTESTRING
52+import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
53+-- import qualified "unix-bytestring" System.Posix.IO.ByteString.Lazy as PIO
54+import System.Posix.IO as PIO (openFd, closeFd, -- append, exclusive, noctty, nonBlock, trunc,
55+ OpenFileFlags(..), OpenMode(ReadOnly, WriteOnly))
56+-- (fromS,toS) = (BS.pack, BS.unpack)
57+(fromS,toS) = (BSS.pack, BSS.unpack)
58+fromBS = id
59+readit fd n = PIO.fdRead fd n
60+#else
61+import qualified System.Posix.IO as PIO
62+(toS,fromS) = (id,id)
63+fromBS = BSS.unpack
64+readit fd n = do (s,_) <- PIO.fdRead fd n
65+ return (BSS.pack s)
66+#endif
67+-- readit :: Fd -> Int -> IO BS.ByteString
68+readit :: Fd -> ByteCount -> IO BSS.ByteString
69+
70+----------------------------------------------------------------------------------------------------
71+
72+-- The msg header consists of just a length field represented as a Word32
73+sizeof_header = 4
74+
75+fileFlags =
76+ PIO.OpenFileFlags {
77+ PIO.append = False,
78+ PIO.exclusive = False,
79+ PIO.noctty = False,
80+ PIO.nonBlock = True,
81+ -- In nonblocking mode opening for read will always succeed and
82+ -- opening for write must happen second.
83+ PIO.trunc = False
84+ }
85+-- NOTE:
86+-- "The only open file status flags that can be meaningfully applied
87+-- to a pipe or FIFO are O_NONBLOCK and O_ASYNC. "
88+
89+
90+mkTransport :: IO Transport
91+mkTransport = do
92+ uid <- randomIO :: IO Word64
93+ lock <- newMVar ()
94+ let filename = "/tmp/pipe_"++show uid
95+ createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
96+
97+ return Transport
98+ { newConnectionWith = \ _ -> do
99+ -- Here we protect from blocking other threads by running on a separate (OS) thread:
100+ -- Opening the file on the reader side should always succeed:
101+ mv <- onOSThread$ tryUntilNoIOErr $
102+-- spinTillThere filename
103+ PIO.openFd filename PIO.ReadOnly Nothing fileFlags
104+
105+ return (mkSourceAddr filename,
106+ mkTargetEnd mv lock)
107+ , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
108+ , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
109+ , closeTransport = removeFile filename
110+ }
111+ where
112+ mkSourceAddr :: String -> SourceAddr
113+ mkSourceAddr filename = SourceAddr
114+ { connectWith = \_ -> mkSourceEnd filename
115+ , serialize = BS.pack filename
116+ }
117+
118+ mkSourceEnd :: String -> IO SourceEnd
119+ mkSourceEnd filename = do
120+ -- Initiate but do not block on file opening:
121+ mv <- onOSThread$ tryUntilNoIOErr $
122+ -- The reader must connect first, the writer here spins with backoff.
123+ PIO.openFd filename PIO.WriteOnly Nothing fileFlags
124+
125+ return $
126+ -- Write to the named pipe. If the message is less than
127+ -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
128+ -- we have to do something more sophisticated.
129+ SourceEnd
130+ { send = \bsls -> do
131+ -- ThreadSafe: This may happen on multiple processes/threads:
132+ let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bsls
133+ when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
134+ error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
135+ -- Otherwise it's just a simple write:
136+ -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
137+
138+ let hdrbss :: BSS.ByteString
139+#ifdef CEREAL
140+ hdrbss = encode msgsize
141+#else
142+-- hdrbss = BS.fromChunks[encode msgsize]
143+ [hdrbss] = BS.toChunks (encode msgsize)
144+#endif
145+
146+-- let finalmsg = BS.concat (encode msgsize : bss)
147+ let finalmsg = BSS.concat (hdrbss : concatMap BS.toChunks bsls)
148+
149+ fd <- readMVar mv -- Synchronize with file opening.
150+ ----------------------------------------
151+ cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
152+ unless (fromIntegral cnt == BSS.length finalmsg) $
153+ error$ "Failed to write message in one go, length: "++ show (BSS.length finalmsg)
154+ ----------------------------------------
155+ return ()
156+ , closeSourceEnd = do
157+ fd <- readMVar mv
158+ PIO.closeFd fd
159+ }
160+
161+ mkTargetEnd :: MVar Fd -> MVar () -> TargetEnd
162+ mkTargetEnd mv lock = TargetEnd
163+ { receive = do
164+ fd <- readMVar mv -- Make sure Fd is there before
165+
166+ -- This should only happen on a single process. But it may
167+ -- happen on multiple threads so we grab a lock.
168+ takeMVar lock
169+
170+ let spinread :: Int -> IO BSS.ByteString
171+ spinread desired = do
172+
173+ bs <- tryUntilNoIOErr$
174+ readit fd (fromIntegral desired)
175+
176+ case BSS.length bs of
177+ n | n == desired -> return bs
178+ 0 -> do threadDelay (10*1000)
179+ spinread desired
180+ l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
181+ show desired ++" bytes) got "++ show l ++ " bytes"
182+
183+ hdr <- spinread sizeof_header
184+ let -- decoded :: BSS.ByteString
185+#ifdef CEREAL
186+-- decoded = decode (BSS.concat$ BS.toChunks hdr)
187+ decoded = decode hdr
188+#else
189+-- decoded = decode hdr
190+ decoded = decode (BS.fromChunks [hdr])
191+#endif
192+ payload <- case decoded of
193+ Left err -> error$ "ERROR: "++ err
194+ Right size -> spinread (fromIntegral (size::Word32))
195+ putMVar lock ()
196+ return [BS.fromChunks [payload]] -- How terribly listy.
197+ , closeTargetEnd = do
198+ fd <- readMVar mv
199+ PIO.closeFd fd
200+ }
201+
202+spinTillThere :: String -> IO ()
203+spinTillThere filename = mkBackoff >>= loop
204+ where
205+ loop bkoff = do b <- doesFileExist filename
206+ unless b $ do bkoff; loop bkoff
207+
208+mkBackoff :: IO (IO ())
209+mkBackoff =
210+ do tref <- newIORef 1
211+ return$ do t <- readIORef tref
212+ writeIORef tref (min maxwait (2 * t))
213+ threadDelay t
214+ where
215+ maxwait = 50 * 1000
216+
217+tryUntilNoIOErr :: IO a -> IO a
218+tryUntilNoIOErr action = mkBackoff >>= loop
219+ where
220+ loop bkoff =
221+ handle (\ (e :: IOException) ->
222+ do bkoff
223+-- BSS.hPutStr stderr$ BSS.pack$ " got IO err: " ++ show e
224+ -- case ioeGetHandle e of
225+ -- Nothing -> BSS.hPutStrLn stderr$ BSS.pack$ " no hndl io err."
226+ -- Just x -> BSS.hPutStrLn stderr$ BSS.pack$ " HNDL on io err!" ++ show x
227+ loop bkoff) $
228+ action
229+
230+-- Execute an action on its own OS thread. Return an MVar to synchronize on.
231+onOSThread :: IO a -> IO (MVar a)
232+onOSThread action = do
233+ mv <- newEmptyMVar
234+ forkOS (action >>= putMVar mv )
235+ return mv