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