@@ -0,0 +1,235 @@
1+{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables, PackageImports #-}
3+----------------------------------------------------------------------------------------------------
5+-- WARNING -- This transport is not yet completed. TODO:
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
11+----------------------------------------------------------------------------------------------------
13+module Network.Transport.Pipes
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)
27+import qualified Data.IntMap as IntMap
28+import qualified Data.ByteString.Char8 as BSS
29+import qualified Data.ByteString.Lazy.Char8 as BS
31+-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:
34+import Data.Serialize (encode,decode) -- Uses strict BS
36+import Data.Binary (encode,decode) -- Uses lazy BS
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)
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)
59+readit fd n = PIO.fdRead fd n
61+import qualified System.Posix.IO as PIO
64+readit fd n = do (s,_) <- PIO.fdRead fd n
67+-- readit :: Fd -> Int -> IO BS.ByteString
68+readit :: Fd -> ByteCount -> IO BSS.ByteString
70+----------------------------------------------------------------------------------------------------
72+-- The msg header consists of just a length field represented as a Word32
78+ PIO.exclusive = False,
81+ -- In nonblocking mode opening for read will always succeed and
82+ -- opening for write must happen second.
86+-- "The only open file status flags that can be meaningfully applied
87+-- to a pipe or FIFO are O_NONBLOCK and O_ASYNC. "
90+mkTransport :: IO Transport
92+ uid <- randomIO :: IO Word64
94+ let filename = "/tmp/pipe_"++show uid
95+ createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
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
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
112+ mkSourceAddr :: String -> SourceAddr
113+ mkSourceAddr filename = SourceAddr
114+ { connectWith = \_ -> mkSourceEnd filename
115+ , serialize = BS.pack filename
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
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.
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:
138+ let hdrbss :: BSS.ByteString
140+ hdrbss = encode msgsize
142+-- hdrbss = BS.fromChunks[encode msgsize]
143+ [hdrbss] = BS.toChunks (encode msgsize)
146+-- let finalmsg = BS.concat (encode msgsize : bss)
147+ let finalmsg = BSS.concat (hdrbss : concatMap BS.toChunks bsls)
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+ ----------------------------------------
156+ , closeSourceEnd = do
161+ mkTargetEnd :: MVar Fd -> MVar () -> TargetEnd
162+ mkTargetEnd mv lock = TargetEnd
164+ fd <- readMVar mv -- Make sure Fd is there before
166+ -- This should only happen on a single process. But it may
167+ -- happen on multiple threads so we grab a lock.
170+ let spinread :: Int -> IO BSS.ByteString
171+ spinread desired = do
173+ bs <- tryUntilNoIOErr$
174+ readit fd (fromIntegral desired)
176+ case BSS.length bs of
177+ n | n == desired -> return bs
178+ 0 -> do threadDelay (10*1000)
180+ l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
181+ show desired ++" bytes) got "++ show l ++ " bytes"
183+ hdr <- spinread sizeof_header
184+ let -- decoded :: BSS.ByteString
186+-- decoded = decode (BSS.concat$ BS.toChunks hdr)
187+ decoded = decode hdr
189+-- decoded = decode hdr
190+ decoded = decode (BS.fromChunks [hdr])
192+ payload <- case decoded of
193+ Left err -> error$ "ERROR: "++ err
194+ Right size -> spinread (fromIntegral (size::Word32))
196+ return [BS.fromChunks [payload]] -- How terribly listy.
197+ , closeTargetEnd = do
202+spinTillThere :: String -> IO ()
203+spinTillThere filename = mkBackoff >>= loop
205+ loop bkoff = do b <- doesFileExist filename
206+ unless b $ do bkoff; loop bkoff
208+mkBackoff :: IO (IO ())
210+ do tref <- newIORef 1
211+ return$ do t <- readIORef tref
212+ writeIORef tref (min maxwait (2 * t))
217+tryUntilNoIOErr :: IO a -> IO a
218+tryUntilNoIOErr action = mkBackoff >>= loop
221+ handle (\ (e :: IOException) ->
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
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
234+ forkOS (action >>= putMVar mv )