cece028 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start making the transport async-exception safe

cece028Edsko de Vries14 years ago

network-transport/network-transport.cabal+4−4

compare with master

unifiedsplit

@@ -23,7 +23,7 @@ Library

23 Network.Transport.Chan,

24 Network.Transport.TCP,

25 Network.Transport.Util

26− extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes

26+ extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ScopedTypeVariables

27 ghc-options: -Wall -fno-warn-unused-do-bind

28 HS-Source-Dirs: src

@@ -39,7 +39,7 @@ Test-Suite TestTCP

39 transformers,

40 ansi-terminal,

41 random

42− extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

42+ extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ScopedTypeVariables, ExistentialQuantification, FlexibleInstances, OverlappingInstances

43 ghc-options: -Wall -fno-warn-unused-do-bind -threaded -rtsopts -with-rtsopts=-N

44 HS-Source-Dirs: tests src

@@ -54,7 +54,7 @@ Test-Suite TestMulticastInMemory

54 transformers,

55 ansi-terminal,

56 random

57− extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

57+ extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ScopedTypeVariables, ExistentialQuantification, FlexibleInstances, OverlappingInstances

58 ghc-options: -Wall -fno-warn-unused-do-bind

59 HS-Source-Dirs: tests src

@@ -69,6 +69,6 @@ Test-Suite TestInMemory

69 transformers,

70 ansi-terminal,

71 random

72− extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

72+ extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, RankNTypes, ScopedTypeVariables, ExistentialQuantification, FlexibleInstances, OverlappingInstances

73 ghc-options: -Wall -fno-warn-unused-do-bind

74 HS-Source-Dirs: tests src

network-transport/src/Network/Transport/Internal.hs+24

@@ -10,6 +10,7 @@ module Network.Transport.Internal ( -- * Encoders/decoders

10 , tryIO

11 , tryToEnum

12 , timeoutMaybe

13+ , asyncWhenCancelled

13 -- * Replicated functionality from "base"

14 -- * Replicated functionality from "base"

14 , void

15 , void

15 , forkIOWithUnmask

16 , forkIOWithUnmask

@@ -28,13 +29,17 @@ import qualified Data.ByteString.Internal as BSI ( unsafeCreate

28 , inlinePerformIO)

29 , inlinePerformIO)

29import Control.Monad.IO.Class (MonadIO, liftIO)

30import Control.Monad.IO.Class (MonadIO, liftIO)

30import Control.Exception ( IOException

31import Control.Exception ( IOException

32+ , SomeException

33+ , AsyncException

31 , Exception

34 , Exception

32 , catch

35 , catch

33 , try

36 , try

34 , throw

37 , throw

35 , throwIO

38 , throwIO

39+ , mask_

36 )

40 )

37import Control.Concurrent (ThreadId, forkIO)

41import Control.Concurrent (ThreadId, forkIO)

42+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)

38import GHC.IO (unsafeUnmask)

43import GHC.IO (unsafeUnmask)

39import System.Timeout (timeout)

44import System.Timeout (timeout)

40--import Control.Concurrent (myThreadId)

45--import Control.Concurrent (myThreadId)

@@ -124,3 +129,22 @@ timeoutMaybe (Just n) e f = do

124 case ma of

129 case ma of

125 Nothing -> throwIO e

130 Nothing -> throwIO e

126 Just a -> return a

131 Just a -> return a

132+

133+-- | @asyncWhenCancelled g f@ runs f in a separate thread and waits for it

134+-- to complete. If f throws an exception we catch it and rethrow it in the

135+-- current thread. If the current thread is interrupted before f completes,

136+-- we run the specified clean up handler (if f throws an exception we assume

137+-- that no cleanup is necessary).

138+asyncWhenCancelled :: forall a. (a -> IO ()) -> IO a -> IO a

139+asyncWhenCancelled g f = mask_ $ do

140+ mvar <- newEmptyMVar

141+ forkIO $ try f >>= putMVar mvar

142+ -- takeMVar is interruptible (even inside a mask_)

143+ catch (takeMVar mvar) (exceptionHandler mvar) >>= either throwIO return

144+ where

145+ exceptionHandler :: MVar (Either SomeException a)

146+ -> AsyncException

147+ -> IO (Either SomeException a)

148+ exceptionHandler mvar ex = do

149+ forkIO $ takeMVar mvar >>= either (const $ return ()) g

150+ throwIO ex

network-transport/src/Network/Transport/TCP.hs+29−27

@@ -43,6 +43,7 @@ import Network.Transport.Internal ( encodeInt32

43 , tryToEnum

44 , void

45 , timeoutMaybe

46+ , asyncWhenCancelled

46 )

47 )

47import qualified Network.Socket as N ( HostName

48import qualified Network.Socket as N ( HostName

48 , ServiceName

49 , ServiceName

@@ -525,36 +526,37 @@ apiConnect :: TCPParameters -- ^ Parameters

525apiConnect params ourEndPoint theirAddress _reliability hints =

526apiConnect params ourEndPoint theirAddress _reliability hints =

526 if localAddress ourEndPoint == theirAddress

527 if localAddress ourEndPoint == theirAddress

527 then connectToSelf ourEndPoint

528 then connectToSelf ourEndPoint

528− else try $ do

529+ else try . asyncWhenCancelled close $ do

529− resetIfBroken ourEndPoint theirAddress

530+ resetIfBroken ourEndPoint theirAddress

530− (theirEndPoint, connId) <-

531+ (theirEndPoint, connId) <-

531− requestConnectionTo params ourEndPoint theirAddress hints

532+ requestConnectionTo params ourEndPoint theirAddress hints

532− -- connAlive can be an IORef rather than an MVar because it is protected

533+ -- connAlive can be an IORef rather than an MVar because it is protected

533− -- by the remoteState MVar. We don't need the overhead of locking twice.

534+ -- by the remoteState MVar. We don't need the overhead of locking twice.

534− connAlive <- newIORef True

535+ connAlive <- newIORef True

535− return Connection

536+ return Connection

536− { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive

537+ { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive

537− , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive

538+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive

538− }

539+ }

539

540

540-- | Close a connection

541-- | Close a connection

541apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()

542apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()

542−apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do

543+apiClose (ourEndPoint, theirEndPoint) connId connAlive =

544+ void . tryIO . asyncWhenCancelled return $ do

543− modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

545+ modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

544− { caseValid = \vst -> do

546+ { caseValid = \vst -> do

545− alive <- readIORef connAlive

547+ alive <- readIORef connAlive

546− if alive

548+ if alive

547− then do

549+ then do

548− writeIORef connAlive False

550+ writeIORef connAlive False

549− sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]

551+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]

550− return ( RemoteEndPointValid

552+ return ( RemoteEndPointValid

551− . (remoteOutgoing ^: (\x -> x - 1))

553+ . (remoteOutgoing ^: (\x -> x - 1))

552− $ vst

554+ $ vst

553− )

555+ )

554− else

556+ else

555− return (RemoteEndPointValid vst)

557+ return (RemoteEndPointValid vst)

556− }

558+ }

557− closeIfUnused (ourEndPoint, theirEndPoint)

559+ closeIfUnused (ourEndPoint, theirEndPoint)

558

560

559-- | Send data across a connection

561-- | Send data across a connection

560apiSend :: EndPointPair -- ^ Local and remote endpoint

562apiSend :: EndPointPair -- ^ Local and remote endpoint

network-transport/tests/TestTCP.hs+1−1

@@ -775,7 +775,7 @@ main = do

775 ]

776 -- Run the generic tests even if the TCP specific tests failed..

777 testTransport (either (Left . show) (Right) <$> nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters)

778− -- ..but if the generic tests pass, still fail if the specific tests did

778+ -- ..but if the generic tests pass, still fail if the specific tests did not

779 case tcpResult of

780 Left err -> throwIO err

781 Right () -> return ()

network-transport/tests/TestTransport.hs+62−51

@@ -4,12 +4,13 @@ module TestTransport where

4import Prelude hiding (catch, (>>=), (>>), return, fail)

5import TestAuxiliary (forkTry, runTests, trySome, randomThreadDelay)

6import Control.Concurrent (forkIO, killThread, yield)

7−import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar, tryTakeMVar)

7+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar, tryTakeMVar, modifyMVar_, newMVar)

8−import Control.Exception (evaluate, throw, throwIO)

8+import Control.Exception (evaluate, throw, throwIO, bracket)

9−import Control.Monad (replicateM, replicateM_, when, guard, forM, forM_, unless)

9+import Control.Monad (replicateM, replicateM_, when, guard, forM_, unless)

10import Control.Monad.Error ()

11+import Control.Applicative ((<$>))

11import Network.Transport

12import Network.Transport

12−import Network.Transport.Internal (tlog)

13+import Network.Transport.Internal (tlog, tryIO, timeoutMaybe)

13import Network.Transport.Util (spawn)

14import Network.Transport.Util (spawn)

14import System.Random (randomIO)

15import System.Random (randomIO)

15import Data.ByteString (ByteString)

16import Data.ByteString (ByteString)

@@ -271,30 +272,37 @@ testCloseOneDirection transport numPings = do

271

272

272 mapM_ takeMVar [doneA, doneB]

273 mapM_ takeMVar [doneA, doneB]

273

274

274−-- | Collect a given number of events and order them by connection ID

275+-- | Collect events and order them by connection ID

275−collect :: EndPoint -> Int -> IO [(ConnectionId, [[ByteString]])]

276+collect :: EndPoint -> Maybe Int -> Maybe Int -> IO [(ConnectionId, [[ByteString]])]

276−collect endPoint numEvents = go numEvents Map.empty Map.empty

277+collect endPoint maxEvents timeout = go maxEvents Map.empty Map.empty

277 where

278 where

278 -- TODO: for more serious use of this function we'd need to make these arguments strict

279 -- TODO: for more serious use of this function we'd need to make these arguments strict

279− go 0 open closed = if Map.null open

280+ go (Just 0) open closed = finish open closed

280− then return . Map.toList . Map.map reverse $ closed

281− else fail "Open connections"

282 go n open closed = do

281 go n open closed = do

283− event <- receive endPoint

282+ mEvent <- tryIO . timeoutMaybe timeout (userError "timeout") $ receive endPoint

284− case event of

283+ case mEvent of

284+ Left _ -> finish open closed

285+ Right event -> do

286+ let n' = (\x -> x - 1) <$> n

287+ case event of

285− ConnectionOpened cid _ _ ->

288+ ConnectionOpened cid _ _ ->

286− go (n - 1) (Map.insert cid [] open) closed

289+ go n' (Map.insert cid [] open) closed

287− ConnectionClosed cid ->

290+ ConnectionClosed cid ->

288− let list = Map.findWithDefault (error "Invalid ConnectionClosed") cid open in

291+ let list = Map.findWithDefault (error "Invalid ConnectionClosed") cid open in

289− go (n - 1) (Map.delete cid open) (Map.insert cid list closed)

292+ go n' (Map.delete cid open) (Map.insert cid list closed)

290− Received cid msg ->

293+ Received cid msg ->

291− go (n - 1) (Map.adjust (msg :) cid open) closed

294+ go n' (Map.adjust (msg :) cid open) closed

292− ReceivedMulticast _ _ ->

295+ ReceivedMulticast _ _ ->

293− fail "Unexpected multicast"

296+ fail "Unexpected multicast"

294− ErrorEvent _ ->

297+ ErrorEvent _ ->

295− fail "Unexpected error"

298+ fail "Unexpected error"

296− EndPointClosed ->

299+ EndPointClosed ->

297− fail "Unexpected endpoint closure"

300+ fail "Unexpected endpoint closure"

301+

302+ finish open closed =

303+ if Map.null open

304+ then return . Map.toList . Map.map reverse $ closed

305+ else fail "Open connections"

298

306

299-- | Open connection, close it, then reopen it

307-- | Open connection, close it, then reopen it

300-- (In the TCP transport this means the socket will be closed, then reopened)

308-- (In the TCP transport this means the socket will be closed, then reopened)

@@ -342,7 +350,7 @@ testCloseReopen transport numPings = do

342 Right endpoint <- newEndPoint transport

350 Right endpoint <- newEndPoint transport

343 putMVar addrB (address endpoint)

351 putMVar addrB (address endpoint)

344

352

345− eventss <- collect endpoint (numRepeats * (numPings + 2))

353+ eventss <- collect endpoint (Just (numRepeats * (numPings + 2))) Nothing

346

354

347 forM_ (zip [1 .. numRepeats] eventss) $ \(i, (_, events)) -> do

355 forM_ (zip [1 .. numRepeats] eventss) $ \(i, (_, events)) -> do

348 forM_ (zip [1 .. numPings] events) $ \(j, event) -> do

356 forM_ (zip [1 .. numPings] events) $ \(j, event) -> do

@@ -368,7 +376,7 @@ testParallelConnects transport numPings = do

368 close conn

376 close conn

369

377

370 forkTry $ do

378 forkTry $ do

371− eventss <- collect endpoint (numPings * 4)

379+ eventss <- collect endpoint (Just (numPings * 4)) Nothing

372 -- Check that no pings got sent to the wrong connection

380 -- Check that no pings got sent to the wrong connection

373 forM_ eventss $ \(_, [[ping1], [ping2]]) ->

381 forM_ eventss $ \(_, [[ping1], [ping2]]) ->

374 guard (ping1 == ping2)

382 guard (ping1 == ping2)

@@ -519,7 +527,7 @@ testConnectToSelfTwice transport numPings = do

519 forkTry $ do

527 forkTry $ do

520 tlog $ "reading"

528 tlog $ "reading"

521

529

522− [(_, events1), (_, events2)] <- collect endpoint (2 * (numPings + 2))

530+ [(_, events1), (_, events2)] <- collect endpoint (Just (2 * (numPings + 2))) Nothing

523 True <- return $ events1 == replicate numPings ["pingA"]

531 True <- return $ events1 == replicate numPings ["pingA"]

524 True <- return $ events2 == replicate numPings ["pingB"]

532 True <- return $ events2 == replicate numPings ["pingB"]

525

533

@@ -826,31 +834,34 @@ testKill newTransport numThreads = do

826 Right endpoint1 <- newEndPoint transport1

834 Right endpoint1 <- newEndPoint transport1

827 Right endpoint2 <- newEndPoint transport2

835 Right endpoint2 <- newEndPoint transport2

828

836

829− Right conn <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints

830− ConnectionOpened _ _ _ <- receive endpoint2

831−

832 threads <- replicateM numThreads . forkIO $ do

837 threads <- replicateM numThreads . forkIO $ do

833− randomThreadDelay 10

838+ randomThreadDelay 100

839+ bracket (connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints)

840+ (\(Right conn) -> randomThreadDelay 100 >> close conn)

834− Right () <- send conn ["ping"]

841+ (\(Right conn) -> randomThreadDelay 100 >> do Right () <- send conn ["ping"] ; return ())

835− return ()

842+

836−

843+ numAlive <- newMVar (0 :: Int)

844+

837− -- Kill half of those threads, and wait on the rest

845+ -- Kill half of those threads

838− killerDone <- newEmptyMVar

839− forkIO $ do

840− killed <- forM threads $ \tid -> do

846+ forkIO . forM_ threads $ \tid -> do

841− shouldKill <- randomIO

847+ shouldKill <- randomIO

848+ if shouldKill

842− when shouldKill $ randomThreadDelay 10 >> killThread tid

849+ then randomThreadDelay 600 >> killThread tid

843− return shouldKill

850+ else modifyMVar_ numAlive (return . (+ 1))

844−

851+

845− -- We should receive at least the pings from the threads that we didn't

852+ -- Since it is impossible to predict when the kill exactly happens, we don't

846− -- kill (we might get more, depending on when exactly the kill happens)

853+ -- know how many connects were opened and how many pings were sent. But we

847− forM_ killed $ \wasKilled -> unless wasKilled $ do

854+ -- should not have any open connections (if we do, collect will throw an

848− Received _ ["ping"] <- receive endpoint2

855+ -- error) and we should have at least the number of pings equal to the number

849− return ()

856+ -- of threads we did *not* kill

850−

857+ eventss <- collect endpoint2 Nothing (Just 1000000)

851− putMVar killerDone ()

858+ let actualPings = sum . map (length . snd) $ eventss

859+ expectedPings <- takeMVar numAlive

860+ unless (actualPings >= expectedPings) $

861+ throwIO (userError "Missing pings")

862+

863+-- print (actualPings, expectedPings)

852

864

853− takeMVar killerDone

854

865

855-- | Set up conditions with a high likelyhood of "crossing" (for transports

866-- | Set up conditions with a high likelyhood of "crossing" (for transports

856-- that multiplex lightweight connections across heavyweight connections)

867-- that multiplex lightweight connections across heavyweight connections)