30fc293 · distributed-process

63 PIO.append = False,

64 PIO.exclusive = False,

65 PIO.noctty = False,

66−-- PIO.nonBlock = False,

67 PIO.nonBlock = True,

66 PIO.nonBlock = True,

68 -- In nonblocking mode opening for read will always succeed and

67 -- In nonblocking mode opening for read will always succeed and

69 -- opening for write must happen second.

68 -- opening for write must happen second.

81 let filename = "/tmp/pipe_"++show uid

80 let filename = "/tmp/pipe_"++show uid

82 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode

81 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode

83

82

84− dbgprint1$ " Created pipe at location: "++ filename

85−

86 return Transport

83 return Transport

87 { newConnectionWith = \ _ -> do

84 { newConnectionWith = \ _ -> do

88− dbgprint1$ " Creating new connection"

89 return (mkSourceAddr filename,

85 return (mkSourceAddr filename,

90 mkTargetEnd filename lock)

86 mkTargetEnd filename lock)

91 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"

87 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"

104 mkSourceEnd :: String -> IO SourceEnd

100 mkSourceEnd :: String -> IO SourceEnd

105 mkSourceEnd filename = do

101 mkSourceEnd filename = do

106 -- Initiate but do not block on file opening:

102 -- Initiate but do not block on file opening:

107− -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.

108− --

109−

110− -- All THREE of the below options were observed to work on a simple demo:

111

103

112− -- OPTION (1)

113 -- Here we protect from blocking other threads by running on a separate (OS) thread:

104 -- Here we protect from blocking other threads by running on a separate (OS) thread:

114−-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags

115− dbgprint1$ "About to try opening writing end:"

116 mv <- onOSThread$ tryUntilNoIOErr $

105 mv <- onOSThread$ tryUntilNoIOErr $

106+ -- The reader must connect first, the writer here spins with backoff.

117 PIO.openFd filename PIO.WriteOnly Nothing fileFlags

107 PIO.openFd filename PIO.WriteOnly Nothing fileFlags

118−-- fd <- takeMVar mv

119−

120− -- OPTION (2) / (3)

121−-- spinTillThere filename

122−

123− -- The reader must connect first, the writer here spins with backoff:

124−-- fd <- PIO.openFd filename PIO.WriteOnly Nothing fileFlags

125−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags

126− dbgprint1$ "GOT WRITING END OPEN ... "

127

108

128 return $

109 return $

129 -- Write to the named pipe. If the message is less than

110 -- Write to the named pipe. If the message is less than

131 -- we have to do something more sophisticated.

112 -- we have to do something more sophisticated.

132 SourceEnd

113 SourceEnd

133 { send = \bss -> do

114 { send = \bss -> do

134− dbgprint1$ "Sending.. "++ show bss

135−

136− -- This may happen on multiple processes/threads:

115+ -- ThreadSafe: This may happen on multiple processes/threads:

137 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss

116 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss

138 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code

117 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code

139 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."

118 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."

141 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:

120 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:

142-- let finalmsg = BS.concat (encode msgsize : bss)

121-- let finalmsg = BS.concat (encode msgsize : bss)

143 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)

122 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)

144− dbgprint1$ " Final send msg: " ++ show finalmsg

123+

145−

146− -- OPTION 2: Speculative file opening, plus this synchronnization:

147− fd <- readMVar mv

124+ fd <- readMVar mv -- Synchronnize with file opening.

148 ----------------------------------------

125 ----------------------------------------

149 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

126 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

150 unless (fromIntegral cnt == BS.length finalmsg) $

127 unless (fromIntegral cnt == BS.length finalmsg) $

151 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)

128 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)

152 ----------------------------------------

129 ----------------------------------------

153−

154 return ()

130 return ()

155 }

131 }

156

132

157 mkTargetEnd :: String -> MVar () -> TargetEnd

133 mkTargetEnd :: String -> MVar () -> TargetEnd

158 mkTargetEnd filename lock = TargetEnd

134 mkTargetEnd filename lock = TargetEnd

159 { receive = do

135 { receive = do

160− dbgprint2$ "Begin receive action..."

161 -- This should only happen on a single process. But it may

136 -- This should only happen on a single process. But it may

162 -- happen on multiple threads so we grab a lock.

137 -- happen on multiple threads so we grab a lock.

163 takeMVar lock

138 takeMVar lock

164− dbgprint2$ " (got lock)"

165

139

166 spinTillThere filename

140 spinTillThere filename

167 -- Opening the file on the reader side should always succeed:

141 -- Opening the file on the reader side should always succeed:

168 fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags

142 fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags

169−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags

170−-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags

171− dbgprint2$ " spawned thread for open..."

172−-- let dummy 0 acc = acc

173−-- dummy n acc = dummy (n-1) (1.1 * acc)

174−-- dbgprint2$ " Silly work: " ++ show (dummy 1000000 0.0001)

175

143

176−-- fd <- readMVar mv

177− let -- oneread :: Int64 -> IO (BS.ByteString, Int64)

178− oneread fd n =

144+ let oneread fd n =

179−-- handle (\e -> do

180−-- dbgprint2$ " Got Exception in read call! "++ show e

181−-- case fromException e of

182−-- Nothing -> dbgprint2$ "It's NOT an IO exception! "

183−-- Just e -> dbgprint2$ "It's an IO exception! "++ show (e :: IOException)

184−-- -- throw (e :: SomeException)

185−-- return undefined

186−-- ) $

187 do bs <- tryUntilNoIOErr$

145 do bs <- tryUntilNoIOErr$

188 readit fd (fromIntegral n)

146 readit fd (fromIntegral n)

189 return (bs, BS.length bs)

147 return (bs, BS.length bs)

190− dbgprint2$ " Attempt read header..."

191

148

192 let spinread :: Int64 -> IO BS.ByteString

149 let spinread :: Int64 -> IO BS.ByteString

193 spinread desired = do

150 spinread desired = do

194−#ifdef DEBUG

151+

195−-- hPutStr stderr "."

152+ bs <- tryUntilNoIOErr$

196− dbgprint2$ " SPINREAD "++ show desired

153+ readit fd (fromIntegral desired)

197−-- BSS.hPutStr stdout (BSS.pack$ " "++show desired)

154+

198−#endif

199−-- fd <- readMVar mv

200− (bytes,len) <- oneread fd desired

201− case len of

155+ case BS.length bs of

202− n | n == desired -> return bytes

156+ n | n == desired -> return bs

203 0 -> do threadDelay (10*1000)

157 0 -> do threadDelay (10*1000)

204 spinread desired

158 spinread desired

205 l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++

159 l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++

206 show desired ++" bytes) got "++ show l ++ " bytes"

160 show desired ++" bytes) got "++ show l ++ " bytes"

207

161

208 hdr <- spinread sizeof_header

162 hdr <- spinread sizeof_header

209− dbgprint2$ " Got header "++ show hdr ++ ", next attempt to read payload:"

210#ifdef CEREAL

163#ifdef CEREAL

211 let decoded = decode (BSS.concat$ BS.toChunks hdr)

164 let decoded = decode (BSS.concat$ BS.toChunks hdr)

212−-- dbgprint2$ " DECODING HDR, bytes "++ show (BS.length hdr) ++ ": "++show hdr

213−-- evaluate decoded

214−-- dbgprint2$ " DONE DECODING HDR"

215#else

165#else

216 let decoded = decode hdr

166 let decoded = decode hdr

217#endif

167#endif

218 payload <- case decoded of

168 payload <- case decoded of

219 Left err -> error$ "ERROR: "++ err

169 Left err -> error$ "ERROR: "++ err

220 Right size -> spinread (fromIntegral (size::Word32))

170 Right size -> spinread (fromIntegral (size::Word32))

221− dbgprint2$ " Got payload "++ show payload

222−

223 putMVar lock ()

171 putMVar lock ()

224 return [payload]

172 return [payload]

225 }

173 }

226

174

227spinTillThere :: String -> IO ()

175spinTillThere :: String -> IO ()

228−spinTillThere filename =

176+spinTillThere filename = mkBackoff >>= loop

229− do dbgprint2$ " Spinning till file present: "++ filename

230− mkBackoff >>= loop

231 where

177 where

232 loop bkoff = do b <- doesFileExist filename

178 loop bkoff = do b <- doesFileExist filename

233 unless b $ do bkoff; loop bkoff

179 unless b $ do bkoff; loop bkoff

234

180

235−

236mkBackoff :: IO (IO ())

181mkBackoff :: IO (IO ())

237mkBackoff =

182mkBackoff =

238 do tref <- newIORef 1

183 do tref <- newIORef 1

248 loop bkoff =

193 loop bkoff =

249 handle (\ (e :: IOException) ->

194 handle (\ (e :: IOException) ->

250 do bkoff

195 do bkoff

251− dbgprint2$ " got IO exn: " ++ show e

252 loop bkoff) $

196 loop bkoff) $

253 action

197 action

254

198

258 mv <- newEmptyMVar

202 mv <- newEmptyMVar

259 forkOS (action >>= putMVar mv )

203 forkOS (action >>= putMVar mv )

260 return mv

204 return mv

261−

262−#ifdef DEBUG

263−dbgprint1 s = do BSS.hPutStrLn stderr (BSS.pack s); hFlush stderr;

264−dbgprint2 s = do BSS.hPutStrLn stdout (BSS.pack s); hFlush stdout;

265−#else

266−dbgprint1 _ = return ()

267−dbgprint2 _ = return ()

268−#endif

269−