95c824e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start introducing simulated network errors

95c824eEdsko de Vries14 years ago

network-transport-tcp/src/Network/Transport/TCP/Mock/Socket.hs+60−14

compare with master

unifiedsplit

@@ -25,6 +25,8 @@ module Network.Transport.TCP.Mock.Socket

25 , defaultHints

26 , defaultProtocol

27 , sOMAXCONN

28+ -- * Debugging API

29+ , scheduleReadAction

28 -- * Internal API

30 -- * Internal API

29 , writeSocket

31 , writeSocket

30 , readSocket

32 , readSocket

@@ -39,7 +41,7 @@ import Control.Category ((>>>))

39import Control.Concurrent.MVar

41import Control.Concurrent.MVar

40import Control.Concurrent.Chan

42import Control.Concurrent.Chan

41import System.IO.Unsafe (unsafePerformIO)

43import System.IO.Unsafe (unsafePerformIO)

42−import Data.Accessor (Accessor, accessor, (^=), (^.))

44+import Data.Accessor (Accessor, accessor, (^=), (^.), (^:))

43import qualified Data.Accessor.Container as DAC (mapMaybe)

45import qualified Data.Accessor.Container as DAC (mapMaybe)

44import System.Timeout (timeout)

46import System.Timeout (timeout)

@@ -105,8 +107,14 @@ data Socket = Socket {

105

107

106data SocketState =

108data SocketState =

107 Uninit

109 Uninit

110+ | BoundSocket {

108− | BoundSocket { socketBacklog :: Chan (Socket, SockAddr, MVar Socket) }

111+ socketBacklog :: Chan (Socket, SockAddr, MVar Socket)

112+ }

113+ | Connected {

109− | Connected { socketPeer :: Maybe Socket, socketBuff :: Chan Message }

114+ socketBuff :: Chan Message

115+ , _socketPeer :: Maybe Socket

116+ , _scheduledReadActions :: [(Int, IO ())]

117+ }

110 | Closed

118 | Closed

111

119

112data Message =

120data Message =

@@ -127,6 +135,12 @@ instance Show AddrInfo where

127instance Show Socket where

135instance Show Socket where

128 show sock = "<<socket " ++ socketDescription sock ++ ">>"

136 show sock = "<<socket " ++ socketDescription sock ++ ">>"

129

137

138+socketPeer :: Accessor SocketState (Maybe Socket)

139+socketPeer = accessor _socketPeer (\peer st -> st { _socketPeer = peer })

140+

141+scheduledReadActions :: Accessor SocketState [(Int, IO ())]

142+scheduledReadActions = accessor _scheduledReadActions (\acts st -> st { _scheduledReadActions = acts })

143+

130getAddrInfo :: Maybe AddrInfo -> Maybe HostName -> Maybe ServiceName -> IO [AddrInfo]

144getAddrInfo :: Maybe AddrInfo -> Maybe HostName -> Maybe ServiceName -> IO [AddrInfo]

131getAddrInfo _ (Just host) (Just port) = do

145getAddrInfo _ (Just host) (Just port) = do

132 validHosts <- get validHostnames

146 validHosts <- get validHostnames

@@ -183,8 +197,9 @@ accept serverSock = do

183 (theirSocket, theirAddress, reply) <- readChan backlog

197 (theirSocket, theirAddress, reply) <- readChan backlog

184 ourBuff <- newChan

198 ourBuff <- newChan

185 ourState <- newMVar Connected {

199 ourState <- newMVar Connected {

200+ socketBuff = ourBuff

186− socketPeer = Just theirSocket

201+ , _socketPeer = Just theirSocket

187− , socketBuff = ourBuff

202+ , _scheduledReadActions = []

188 }

203 }

189 let ourSocket = Socket {

204 let ourSocket = Socket {

190 socketState = ourState

205 socketState = ourState

@@ -225,8 +240,9 @@ connect us serverAddr = do

225 Uninit -> do

240 Uninit -> do

226 buff <- newChan

241 buff <- newChan

227 return Connected {

242 return Connected {

243+ socketBuff = buff

228− socketPeer = Just them

244+ , _socketPeer = Just them

229− , socketBuff = buff

245+ , _scheduledReadActions = []

230 }

246 }

231 _ ->

247 _ ->

232 throwSocketError "connect: already connected"

248 throwSocketError "connect: already connected"

@@ -240,7 +256,7 @@ shutdown sock ShutdownSend = do

240 writeSocket sock CloseSocket

256 writeSocket sock CloseSocket

241 timeoutThrow mvarThreshold $ modifyMVar_ (socketState sock) $ \st -> case st of

257 timeoutThrow mvarThreshold $ modifyMVar_ (socketState sock) $ \st -> case st of

242 Connected {} ->

258 Connected {} ->

243− return (Connected Nothing (socketBuff st))

259+ return (socketPeer ^= Nothing $ st)

244 _ ->

260 _ ->

245 return st

261 return st

246

262

@@ -252,7 +268,7 @@ peerBuffer :: Socket -> IO (Either String (Chan Message))

252peerBuffer sock = do

268peerBuffer sock = do

253 mPeer <- timeoutThrow mvarThreshold $ withMVar (socketState sock) $ \st -> case st of

269 mPeer <- timeoutThrow mvarThreshold $ withMVar (socketState sock) $ \st -> case st of

254 Connected {} ->

270 Connected {} ->

255− return (socketPeer st)

271+ return (st ^. socketPeer)

256 _ ->

272 _ ->

257 return Nothing

273 return Nothing

258 case mPeer of

274 case mPeer of

@@ -277,13 +293,17 @@ writeSocket sock msg = do

277

293

278readSocket :: Socket -> IO (Maybe Word8)

294readSocket :: Socket -> IO (Maybe Word8)

279readSocket sock = do

295readSocket sock = do

280− mBuff <- timeoutThrow mvarThreshold $ withMVar (socketState sock) $ \st -> case st of

296+ mBuff <- timeoutThrow mvarThreshold $ modifyMVar (socketState sock) $ \st -> case st of

281− Connected {} ->

297+ Connected {} -> do

298+ let (later, now) = tick $ st ^. scheduledReadActions

299+ return ( scheduledReadActions ^= later $ st

282− return (Just $ socketBuff st)

300+ , Just (socketBuff st, now)

301+ )

283 _ ->

302 _ ->

284− return Nothing

303+ return (st, Nothing)

285 case mBuff of

304 case mBuff of

286− Just buff -> do

305+ Just (buff, actions) -> do

306+ sequence actions

287 msg <- timeoutThrow readSocketThreshold $ readChan buff

307 msg <- timeoutThrow readSocketThreshold $ readChan buff

288 case msg of

308 case msg of

289 Payload w -> return (Just w)

309 Payload w -> return (Just w)

@@ -295,6 +315,32 @@ readSocket sock = do

295 Nothing ->

315 Nothing ->

296 return Nothing

316 return Nothing

297

317

318+-- | Given a list of scheduled actions, reduce all delays by 1, and return the

319+-- actions that should be executed now.

320+tick :: [(Int, IO ())] -> ([(Int, IO ())], [IO ()])

321+tick = go [] []

322+ where

323+ go later now [] = (reverse later, reverse now)

324+ go later now ((n, action) : actions)

325+ | n == 0 = go later (action : now) actions

326+ | otherwise = go ((n - 1, action) : later) now actions

327+

328+--------------------------------------------------------------------------------

329+-- Debugging API --

330+--------------------------------------------------------------------------------

331+

332+-- | Schedule an action to be executed after /n/ reads on this socket

333+--

334+-- If /n/ is zero we execute the action immediately.

335+scheduleReadAction :: Socket -> Int -> IO () -> IO ()

336+scheduleReadAction _ 0 action = action

337+scheduleReadAction sock n action =

338+ modifyMVar_ (socketState sock) $ \st -> case st of

339+ Connected {} ->

340+ return (scheduledReadActions ^: ((n, action) :) $ st)

341+ _ ->

342+ throwSocketError "scheduleReadAction: socket not connected"

343+

298--------------------------------------------------------------------------------

344--------------------------------------------------------------------------------

299-- Util --

345-- Util --

300--------------------------------------------------------------------------------

346--------------------------------------------------------------------------------

network-transport-tcp/tests/TestQC.hs+69−18

@@ -17,6 +17,7 @@ import Test.QuickCheck

17 , forAllShrink

18 , Property

19 , Arbitrary(arbitrary)

20+ , elements

20 )

21 )

21import Test.QuickCheck.Property (morallyDubiousIOProperty, Result(..), result)

22import Test.QuickCheck.Property (morallyDubiousIOProperty, Result(..), result)

22import Test.HUnit (Assertion, assertFailure)

23import Test.HUnit (Assertion, assertFailure)

@@ -32,23 +33,47 @@ import Data.ByteString.Char8 (pack)

32import qualified Data.ByteString as BSS (concat)

33import qualified Data.ByteString as BSS (concat)

34import Network.Transport

35import Network.Transport

35−import Network.Transport.TCP (createTransport, defaultTCPParameters)

36+import Network.Transport.TCP

37+ ( createTransportExposeInternals

38+ , defaultTCPParameters

39+ , TransportInternals(socketBetween)

40+ )

41+

42+import Network.Transport.TCP.Mock.Socket (scheduleReadAction, sClose)

37--------------------------------------------------------------------------------

44--------------------------------------------------------------------------------

38-- Script infrastructure --

45-- Script infrastructure --

39--------------------------------------------------------------------------------

46--------------------------------------------------------------------------------

48+-- | We randomly generate /scripts/ which are essentially a deep embedding of

49+-- the Transport API. These scripts are then executed and the results compared

50+-- against an abstract interpreter.

41data ScriptCmd =

51data ScriptCmd =

52+ -- | Create a new endpoint

42 NewEndPoint

53 NewEndPoint

54+ -- | @Connect i j@ creates a connection from endpoint @i@ to endpoint @j@,

55+ -- where @i@ and @j@ are indices and refer to the @i@th and @j@th endpoint

56+ -- created by NewEndPoint

43 | Connect Int Int

57 | Connect Int Int

58+ -- | @Close i@ closes the @i@ connection created using 'Connect'. Note that

59+ -- closing a connection does not shift other indices; in other words, in

60+ -- @[Connect 0 0, Close 0, Connect 0 0, Close 0]@ the second 'Close'

61+ -- refers to the first (already closed) connection

44 | Close Int

62 | Close Int

63+ -- | @Send i bs@ sends payload @bs@ on the @i@ connection created

45 | Send Int [ByteString]

64 | Send Int [ByteString]

65+ -- | @BreakAfterReads n i j@ force-closes the socket between endpoints @i@

66+ -- and @j@ after @n@ reads by @i@

67+ --

68+ -- We should have @i /= j@ because the TCP transport does not use sockets

69+ -- for connections from an endpoint to itself

70+ | BreakAfterReads Int Int Int

46 deriving Show

71 deriving Show

48type Script = [ScriptCmd]

73type Script = [ScriptCmd]

50−execScript :: Transport -> Script -> IO (Map Int [Event])

75+execScript :: (Transport, TransportInternals) -> Script -> IO (Map Int [Event])

51−execScript transport script = do

76+execScript (transport, transportInternals) script = do

52 chan <- newChan

77 chan <- newChan

53 runScript chan script

78 runScript chan script

54 collectAll chan

79 collectAll chan

@@ -78,6 +103,10 @@ execScript transport script = do

78 Right () <- send (conns !! connIx) payload

103 Right () <- send (conns !! connIx) payload

79 threadDelay 10000

104 threadDelay 10000

80 go endPoints conns cmds

105 go endPoints conns cmds

106+ go endPoints conns (BreakAfterReads n i j : cmds) = do

107+ sock <- socketBetween transportInternals (address (endPoints !! i)) (address (endPoints !! j))

108+ scheduleReadAction sock n (sClose sock)

109+ go endPoints conns cmds

110

82 forwardTo :: Chan (Maybe (Int, Event)) -> (Int, EndPoint) -> IO ()

111 forwardTo :: Chan (Maybe (Int, Event)) -> (Int, EndPoint) -> IO ()

83 forwardTo chan (ix, endPoint) = go

112 forwardTo chan (ix, endPoint) = go

@@ -136,6 +165,8 @@ verify script = go script []

136 go cmds conns (Map.insert epIx epEvs' evs)

165 go cmds conns (Map.insert epIx epEvs' evs)

137 _ ->

166 _ ->

138 Just $ "Missing (Received " ++ show connId ++ " " ++ show payload ++ ") event in " ++ show epEvs

167 Just $ "Missing (Received " ++ show connId ++ " " ++ show payload ++ ") event in " ++ show epEvs

168+ go (BreakAfterReads n i j : cmds) conns evs =

169+ go cmds conns evs

139

170

140--------------------------------------------------------------------------------

171--------------------------------------------------------------------------------

141-- Script generators --

172-- Script generators --

@@ -223,6 +254,25 @@ script_ConnectSendClose numEndPoints = do

223 isOpen :: Map Int Bool -> Int -> Bool

254 isOpen :: Map Int Bool -> Int -> Bool

224 isOpen conns connIx = connIx `Map.member` conns && conns Map.! connIx

255 isOpen conns connIx = connIx `Map.member` conns && conns Map.! connIx

225

256

257+withErrors :: Int -> Gen Script -> Gen Script

258+withErrors numErrors gen = gen >>= insertError numErrors

259+ where

260+ insertError :: Int -> Script -> Gen Script

261+ insertError _ [] = return []

262+ insertError n (Connect i j : cmds) | i /= j = do

263+ insert <- arbitrary

264+ if insert && n > 0

265+ then do

266+ -- We sometimes want big delays, but usually we want short delays

267+ numReads <- elements (concat (replicate 10 [0 .. 9]) ++ [10 ..99])

268+ return $ Connect i j : BreakAfterReads numReads i j : cmds

269+ else do

270+ cmds' <- insertError (n - 1) cmds

271+ return $ Connect i j : cmds'

272+ insertError n (cmd : cmds) = do

273+ cmds' <- insertError n cmds

274+ return $ cmd : cmds'

275+

226--------------------------------------------------------------------------------

276--------------------------------------------------------------------------------

227-- Individual scripts to test specific bugs --

277-- Individual scripts to test specific bugs --

228--------------------------------------------------------------------------------

278--------------------------------------------------------------------------------

@@ -283,41 +333,42 @@ script_Bug1 = [

283-- Main application driver --

333-- Main application driver --

284--------------------------------------------------------------------------------

334--------------------------------------------------------------------------------

285

335

286−basicTests :: Transport -> Int -> [Test]

336+basicTests :: (Transport, TransportInternals) -> Int -> (Gen Script -> Gen Script) -> [Test]

287−basicTests transport numEndPoints = [

337+basicTests transport numEndPoints trans = [

288− testGen "NewEndPoint" transport (script_NewEndPoint 2)

338+ testGen "NewEndPoint" transport (trans (script_NewEndPoint numEndPoints))

289− , testGen "Connect" transport (script_Connect 2)

339+ , testGen "Connect" transport (trans (script_Connect numEndPoints))

290− , testGen "ConnectClose" transport (script_ConnectClose 2)

340+ , testGen "ConnectClose" transport (trans (script_ConnectClose numEndPoints))

291− , testGen "ConnectSendClose" transport (script_ConnectSendClose 2)

341+ , testGen "ConnectSendClose" transport (trans (script_ConnectSendClose numEndPoints))

292 ]

342 ]

293

343

294−tests :: Transport -> [Test]

344+tests :: (Transport, TransportInternals) -> [Test]

295tests transport = [

345tests transport = [

296 testGroup "Bugs" [

346 testGroup "Bugs" [

297 testOne "Bug1" transport script_Bug1

347 testOne "Bug1" transport script_Bug1

298 ]

348 ]

299− , testGroup "One endpoint, with delays" (basicTests transport 1)

349+ , testGroup "One endpoint, with delays" (basicTests transport 1 id)

300− , testGroup "Two endpoints, with delays" (basicTests transport 2)

350+ , testGroup "Two endpoints, with delays" (basicTests transport 2 id)

301− , testGroup "Three endpoints, with delays" (basicTests transport 3)

351+ , testGroup "Three endpoints, with delays" (basicTests transport 3 id)

352+ , testGroup "Four endpoints, with delay, single error" (basicTests transport 4 (withErrors 1))

302 ]

353 ]

303 where

354 where

304

355

305−testOne :: TestName -> Transport -> Script -> Test

356+testOne :: TestName -> (Transport, TransportInternals) -> Script -> Test

306testOne label transport script = testCase label (testScript transport script)

357testOne label transport script = testCase label (testScript transport script)

307

358

308−testGen :: TestName -> Transport -> Gen Script -> Test

359+testGen :: TestName -> (Transport, TransportInternals) -> Gen Script -> Test

309testGen label transport script = testProperty label (testScriptGen transport script)

360testGen label transport script = testProperty label (testScriptGen transport script)

310

361

311main :: IO ()

362main :: IO ()

312main = do

363main = do

313− Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

364+ Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

314 defaultMain (tests transport)

365 defaultMain (tests transport)

315

366

316--------------------------------------------------------------------------------

367--------------------------------------------------------------------------------

317-- Test infrastructure --

368-- Test infrastructure --

318--------------------------------------------------------------------------------

369--------------------------------------------------------------------------------

319

370

320−testScriptGen :: Transport -> Gen Script -> Property

371+testScriptGen :: (Transport, TransportInternals) -> Gen Script -> Property

321testScriptGen transport scriptGen =

372testScriptGen transport scriptGen =

322 forAll scriptGen $ \script ->

373 forAll scriptGen $ \script ->

323 morallyDubiousIOProperty $ do

374 morallyDubiousIOProperty $ do

@@ -330,7 +381,7 @@ testScriptGen transport scriptGen =

330 , reason = '\n' : err ++ "\nAll events: " ++ show evs

381 , reason = '\n' : err ++ "\nAll events: " ++ show evs

331 }

382 }

332

383

333−testScript :: Transport -> Script -> Assertion

384+testScript :: (Transport, TransportInternals) -> Script -> Assertion

334testScript transport script = do

385testScript transport script = do

335 logShow script

386 logShow script

336 evs <- execScript transport script

387 evs <- execScript transport script