b4da726 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Code cleanup

b4da726Edsko de Vries14 years ago

network-transport/src/Network/Transport/Chan.hs+6−4

compare with master

unifiedsplit

@@ -73,11 +73,13 @@ apiConnect myAddress state theirAddress _reliability _hints = do

73-- | Send a message over a connection

74apiSend :: Chan Event -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())

75apiSend chan conn connAlive msg =

76− modifyMVar connAlive $ \alive -> do

76+ modifyMVar connAlive $ \alive ->

77 if alive

78+ then do

78− then do writeChan chan (Received conn msg)

79+ writeChan chan (Received conn msg)

79− return (alive, Right ())

80+ return (alive, Right ())

81+ else

80− else do return (alive, Left (TransportError SendFailed "Connection closed"))

82+ return (alive, Left (TransportError SendFailed "Connection closed"))

82-- | Close a connection

84-- | Close a connection

83apiClose :: Chan Event -> ConnectionId -> MVar Bool -> IO ()

85apiClose :: Chan Event -> ConnectionId -> MVar Bool -> IO ()

network-transport/src/Network/Transport/TCP.hs+60−65

@@ -29,7 +29,7 @@ module Network.Transport.TCP ( -- * Main API

29 -- $design

30 ) where

32−import Prelude hiding (catch)

32+import Prelude hiding (catch, mapM_)

33import Network.Transport

34import Network.Transport.Internal.TCP ( forkServer

35 , recvWithLength

@@ -97,10 +97,10 @@ import qualified Data.IntMap as IntMap (empty)

97import Data.IntSet (IntSet)

98import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete)

99import Data.Map (Map)

100−import qualified Data.Map as Map (empty, elems)

100+import qualified Data.Map as Map (empty)

101import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

102import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)

103−import Data.Foldable (forM_)

103+import Data.Foldable (forM_, mapM_)

104import System.IO (hPutStrLn, stderr)

105

106-- $design

@@ -358,13 +358,8 @@ data TransportInternals = TransportInternals

358--

359-- TODOs: deal with hints

360createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)

361−createTransport host port = do

361+createTransport host port =

362− result <- createTransportExposeInternals host port

362+ either Left (Right . fst) <$> createTransportExposeInternals host port

363− case result of

364− Left err ->

365− return $ Left err

366− Right (transport, _) ->

367− return $ Right transport

368

363

369-- | You should probably not use this function (used for unit testing only)

364-- | You should probably not use this function (used for unit testing only)

370createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, TransportInternals))

365createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, TransportInternals))

@@ -382,12 +377,14 @@ createTransportExposeInternals host port = do

382 -- http://tangentsoft.net/wskfaq/advanced.html#backlog

377 -- http://tangentsoft.net/wskfaq/advanced.html#backlog

383 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html

378 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html

384 tid <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)

379 tid <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)

380+ return ( Transport

385− return ( Transport { newEndPoint = apiNewEndPoint transport

381+ { newEndPoint = apiNewEndPoint transport

386− , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]

382+ , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]

387− }

383+ }

388− , TransportInternals { transportThread = tid

384+ , TransportInternals

385+ { transportThread = tid

389− , socketBetween = internalSocketBetween transport

386+ , socketBetween = internalSocketBetween transport

390− }

387+ }

391 )

388 )

392 where

389 where

393 terminationHandler :: TCPTransport -> SomeException -> IO ()

390 terminationHandler :: TCPTransport -> SomeException -> IO ()

@@ -405,13 +402,11 @@ apiCloseTransport transport mTransportThread evs = do

405 mTSt <- modifyMVar (transportState transport) $ \st -> case st of

402 mTSt <- modifyMVar (transportState transport) $ \st -> case st of

406 TransportValid vst -> return (TransportClosed, Just vst)

403 TransportValid vst -> return (TransportClosed, Just vst)

407 TransportClosed -> return (TransportClosed, Nothing)

404 TransportClosed -> return (TransportClosed, Nothing)

408− forM_ mTSt $ \tSt -> do

409− mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)

405+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)

410− forM_ mTransportThread $ \tid -> do

411− -- This will invoke the termination handler, which in turn will call

406+ -- This will invoke the termination handler, which in turn will call

412− -- apiCloseTransport again, but at that point the transport is

407+ -- apiCloseTransport again, but then the transport will already be closed and

413− -- already in closed state so at that point we return immediately

408+ -- we won't be passed a transport thread, so we terminate immmediate

414− killThread tid

409+ forM_ mTransportThread killThread

415

410

416-- | Create a new endpoint

411-- | Create a new endpoint

417apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

412apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

@@ -442,9 +437,10 @@ apiConnect ourEndPoint theirAddress _reliability _hints = try $ do

442 -- connAlive can be an IORef rather than an MVar because it is protected by

437 -- connAlive can be an IORef rather than an MVar because it is protected by

443 -- the remoteState MVar. We don't need the overhead of locking twice.

438 -- the remoteState MVar. We don't need the overhead of locking twice.

444 connAlive <- newIORef True

439 connAlive <- newIORef True

440+ return Connection

445− return $ Connection { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive

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

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

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

447− }

443+ }

448

444

449-- | Close a connection

445-- | Close a connection

450--

446--

@@ -468,9 +464,9 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do

468 else

464 else

469 return (RemoteEndPointValid vst))

465 return (RemoteEndPointValid vst))

470 -- RemoteEndPointClosing

466 -- RemoteEndPointClosing

471− (\(resolved, vst) -> return $ RemoteEndPointClosing resolved vst)

467+ (return . uncurry RemoteEndPointClosing)

472 -- RemoteEndPoinClosed

468 -- RemoteEndPoinClosed

473− (return $ RemoteEndPointClosed)

469+ (return RemoteEndPointClosed)

474 closeIfUnused (ourEndPoint, theirEndPoint)

470 closeIfUnused (ourEndPoint, theirEndPoint)

475

471

476-- | Send data across a connection

472-- | Send data across a connection

@@ -478,8 +474,8 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do

478-- RELY: The endpoint must not be in invalid state.

474-- RELY: The endpoint must not be in invalid state.

479-- GUARANTEE: The state of the remote endpoint will not be changed.

475-- GUARANTEE: The state of the remote endpoint will not be changed.

480apiSend :: EndPointPair -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())

476apiSend :: EndPointPair -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())

481−apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = do

477+apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

482− join <$> (try . mapExceptionIO sendFailed $ modifyRemoteState (ourEndPoint, theirEndPoint)

478+ join <$> (try . mapExceptionIO sendFailed $ withRemoteState (ourEndPoint, theirEndPoint)

483 -- RemoteEndPointInvalid

479 -- RemoteEndPointInvalid

484 (\_ -> fail "apiSend RELY violation")

480 (\_ -> fail "apiSend RELY violation")

485 -- RemoteEndPointValid

481 -- RemoteEndPointValid

@@ -488,13 +484,13 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = do

488 if alive

484 if alive

489 then do

485 then do

490 sendOn vst (encodeInt32 connId : prependLength payload)

486 sendOn vst (encodeInt32 connId : prependLength payload)

491− return (RemoteEndPointValid vst, Right $ ())

487+ return . Right $ ()

492 else

488 else

493− return (RemoteEndPointValid vst, Left $ TransportError SendClosed "Connection closed"))

489+ return . Left $ TransportError SendClosed "Connection closed")

494 -- RemoteEndPointClosing

490 -- RemoteEndPointClosing

495− (\(resolved, vst) -> return (RemoteEndPointClosing resolved vst, Left $ TransportError SendClosed "Connection lost"))

491+ (return . Left . const (TransportError SendClosed "Connection lost"))

496 -- RemoteEndPointClosed

492 -- RemoteEndPointClosed

497− (return (RemoteEndPointClosed, Left $ TransportError SendClosed "Connection lost")))

493+ (return . Left $ TransportError SendClosed "Connection lost"))

498 where

494 where

499 sendFailed :: IOException -> TransportError SendErrorCode

495 sendFailed :: IOException -> TransportError SendErrorCode

500 sendFailed = TransportError SendFailed . show

496 sendFailed = TransportError SendFailed . show

@@ -516,13 +512,13 @@ apiCloseEndPoint transport evs ourEndPoint = do

516 return (LocalEndPointClosed, Nothing)

512 return (LocalEndPointClosed, Nothing)

517 forM_ mOurState $ \vst -> do

513 forM_ mOurState $ \vst -> do

518 -- Close all endpoints and kill all threads

514 -- Close all endpoints and kill all threads

519− forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket

515+ forM_ (vst ^. localConnections) tryCloseRemoteSocket

520 forM_ (vst ^. internalThreads) killThread

516 forM_ (vst ^. internalThreads) killThread

521 forM_ evs $ writeChan (localChannel ourEndPoint)

517 forM_ evs $ writeChan (localChannel ourEndPoint)

522 where

518 where

523 -- Close the remote socket and return the set of all incoming connections

519 -- Close the remote socket and return the set of all incoming connections

524 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

520 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

525− tryCloseRemoteSocket theirEndPoint = do

521+ tryCloseRemoteSocket theirEndPoint =

526 -- We make an attempt to close the connection nicely (by sending a CloseSocket first)

522 -- We make an attempt to close the connection nicely (by sending a CloseSocket first)

527 modifyMVar_ (remoteState theirEndPoint) $ \st ->

523 modifyMVar_ (remoteState theirEndPoint) $ \st ->

528 case st of

524 case st of

@@ -560,7 +556,7 @@ connectToSelf ourEndPoint = do

560 ourChan = localChannel ourEndPoint

556 ourChan = localChannel ourEndPoint

561

557

562 selfSend :: MVar Bool -> ConnectionId -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())

558 selfSend :: MVar Bool -> ConnectionId -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())

563− selfSend connAlive connId msg = do

559+ selfSend connAlive connId msg =

564 modifyMVar connAlive $ \alive ->

560 modifyMVar connAlive $ \alive ->

565 if alive

561 if alive

566 then do

562 then do

@@ -570,7 +566,7 @@ connectToSelf ourEndPoint = do

570 return (alive, Left (TransportError SendFailed "Connection closed"))

566 return (alive, Left (TransportError SendFailed "Connection closed"))

571

567

572 selfClose :: MVar Bool -> ConnectionId -> IO ()

568 selfClose :: MVar Bool -> ConnectionId -> IO ()

573− selfClose connAlive connId = do

569+ selfClose connAlive connId =

574 modifyMVar_ connAlive $ \alive -> do

570 modifyMVar_ connAlive $ \alive -> do

575 when alive $ writeChan ourChan (ConnectionClosed connId)

571 when alive $ writeChan ourChan (ConnectionClosed connId)

576 return False

572 return False

@@ -585,23 +581,23 @@ internalSocketBetween :: TCPTransport -- ^ Transport

585 -> EndPointAddress -- ^ Remote endpoint

581 -> EndPointAddress -- ^ Remote endpoint

586 -> IO (Either String N.Socket)

582 -> IO (Either String N.Socket)

587internalSocketBetween transport ourAddress theirAddress = runErrorT $ do

583internalSocketBetween transport ourAddress theirAddress = runErrorT $ do

588− ourEndPoint <- ErrorT $ do

584+ ourEndPoint <- ErrorT $

589 withMVar (transportState transport) $ \st -> case st of

585 withMVar (transportState transport) $ \st -> case st of

590 TransportClosed ->

586 TransportClosed ->

591 return . Left $ "Transport closed"

587 return . Left $ "Transport closed"

592− TransportValid vst -> do

588+ TransportValid vst ->

593 case vst ^. localEndPointAt ourAddress of

589 case vst ^. localEndPointAt ourAddress of

594 Nothing -> return . Left $ "Local endpoint not found"

590 Nothing -> return . Left $ "Local endpoint not found"

595 Just ep -> return . Right $ ep

591 Just ep -> return . Right $ ep

596− theirEndPoint <- ErrorT $ do

592+ theirEndPoint <- ErrorT $

597 withMVar (localState ourEndPoint) $ \st -> case st of

593 withMVar (localState ourEndPoint) $ \st -> case st of

598 LocalEndPointClosed ->

594 LocalEndPointClosed ->

599 return . Left $ "Local endpoint closed"

595 return . Left $ "Local endpoint closed"

600− LocalEndPointValid vst -> do

596+ LocalEndPointValid vst ->

601 case vst ^. localConnectionTo theirAddress of

597 case vst ^. localConnectionTo theirAddress of

602 Nothing -> return . Left $ "Remote endpoint not found"

598 Nothing -> return . Left $ "Remote endpoint not found"

603 Just ep -> return . Right $ ep

599 Just ep -> return . Right $ ep

604− ErrorT $ do

600+ ErrorT $

605 withMVar (remoteState theirEndPoint) $ \st -> case st of

601 withMVar (remoteState theirEndPoint) $ \st -> case st of

606 RemoteEndPointValid vst ->

602 RemoteEndPointValid vst ->

607 return . Right $ remoteSocket vst

603 return . Right $ remoteSocket vst

@@ -679,7 +675,7 @@ requestConnectionTo ourEndPoint theirAddress = go

679 -- endpoint state, and might have changed in the meantime, these changes

675 -- endpoint state, and might have changed in the meantime, these changes

680 -- won't matter.

676 -- won't matter.

681 case endPointStateSnapshot of

677 case endPointStateSnapshot of

682− RemoteEndPointInvalid err -> do

678+ RemoteEndPointInvalid err ->

683 throw err

679 throw err

684

680

685 RemoteEndPointClosing resolved _ ->

681 RemoteEndPointClosing resolved _ ->

@@ -687,7 +683,7 @@ requestConnectionTo ourEndPoint theirAddress = go

687 -- this is resolved and we then try again

683 -- this is resolved and we then try again

688 readMVar resolved >> go

684 readMVar resolved >> go

689

685

690− RemoteEndPointClosed -> do

686+ RemoteEndPointClosed ->

691 -- EndPointClosed indicates that a concurrent thread was in the

687 -- EndPointClosed indicates that a concurrent thread was in the

692 -- process of closing the TCP connection to the remote endpoint when

688 -- process of closing the TCP connection to the remote endpoint when

693 -- we obtained a reference to it. By INV-CLOSE we can assume that the

689 -- we obtained a reference to it. By INV-CLOSE we can assume that the

@@ -705,7 +701,7 @@ requestConnectionTo ourEndPoint theirAddress = go

705 case decodeInt32 . BS.concat $ reply of

701 case decodeInt32 . BS.concat $ reply of

706 Nothing ->

702 Nothing ->

707 throw (connectFailed $ userError "Invalid integer")

703 throw (connectFailed $ userError "Invalid integer")

708− Just cid -> do

704+ Just cid ->

709 return (theirEndPoint, cid)

705 return (theirEndPoint, cid)

710

706

711 -- If this is a new endpoint, fork a thread to listen for incoming

707 -- If this is a new endpoint, fork a thread to listen for incoming

@@ -750,14 +746,14 @@ requestConnectionTo ourEndPoint theirAddress = go

750setupRemoteEndPoint :: EndPointPair -> IO ()

746setupRemoteEndPoint :: EndPointPair -> IO ()

751setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do

747setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do

752 didAccept <- bracketOnError (socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint))

748 didAccept <- bracketOnError (socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint))

753− onError $ \result -> do

749+ onError $ \result ->

754 case result of

750 case result of

755 Right (sock, ConnectionRequestAccepted) -> do

751 Right (sock, ConnectionRequestAccepted) -> do

756 let vst = ValidRemoteEndPointState

752 let vst = ValidRemoteEndPointState

757 { remoteSocket = sock

753 { remoteSocket = sock

758 , _remoteOutgoing = 0

754 , _remoteOutgoing = 0

759 , _remoteIncoming = IntSet.empty

755 , _remoteIncoming = IntSet.empty

760− , sendOn = \msg -> sendMany sock msg `onException` (tryIO $ N.sClose sock)

756+ , sendOn = \msg -> sendMany sock msg `onException` tryIO (N.sClose sock)

761 }

757 }

762 putMVar theirState (RemoteEndPointValid vst)

758 putMVar theirState (RemoteEndPointValid vst)

763 return True

759 return True

@@ -827,7 +823,7 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do

827 Just r -> return (sock, r)

823 Just r -> return (sock, r)

828 where

824 where

829 createSocket :: N.AddrInfo -> IO N.Socket

825 createSocket :: N.AddrInfo -> IO N.Socket

830− createSocket addr = mapExceptionIO insufficientResources $ do

826+ createSocket addr = mapExceptionIO insufficientResources $

831 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

827 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

832

828

833 invalidAddress, insufficientResources, failed :: IOException -> TransportError ConnectErrorCode

829 invalidAddress, insufficientResources, failed :: IOException -> TransportError ConnectErrorCode

@@ -839,7 +835,7 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do

839--

835--

840-- If the local endpoint is closed, do nothing

836-- If the local endpoint is closed, do nothing

841removeRemoteEndPoint :: EndPointPair -> IO ()

837removeRemoteEndPoint :: EndPointPair -> IO ()

842−removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do

838+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =

843 modifyMVar_ ourState $ \st -> case st of

839 modifyMVar_ ourState $ \st -> case st of

844 LocalEndPointValid vst ->

840 LocalEndPointValid vst ->

845 case vst ^. localConnectionTo theirAddress of

841 case vst ^. localConnectionTo theirAddress of

@@ -859,7 +855,7 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do

859--

855--

860-- Does nothing if the transport is closed

856-- Does nothing if the transport is closed

861removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()

857removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()

862−removeLocalEndPoint transport ourEndPoint = do

858+removeLocalEndPoint transport ourEndPoint =

863 modifyMVar_ (transportState transport) $ \st -> case st of

859 modifyMVar_ (transportState transport) $ \st -> case st of

864 TransportValid vst ->

860 TransportValid vst ->

865 return (TransportValid . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst)

861 return (TransportValid . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst)

@@ -971,7 +967,7 @@ modifyRemoteState :: EndPointPair

971 -> ((MVar (), ValidRemoteEndPointState) -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointClosing

967 -> ((MVar (), ValidRemoteEndPointState) -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointClosing

972 -> IO (RemoteState, a) -- ^ Case for RemoteEndPointClosed

968 -> IO (RemoteState, a) -- ^ Case for RemoteEndPointClosed

973 -> IO a

969 -> IO a

974−modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do

970+modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =

975 mask $ \restore -> do

971 mask $ \restore -> do

976 st <- takeMVar theirState

972 st <- takeMVar theirState

977 case st of

973 case st of

@@ -981,7 +977,7 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing

981 Right (st', a) -> do

977 Right (st', a) -> do

982 putMVar theirState st'

978 putMVar theirState st'

983 return a

979 return a

984− Left ex -> do

980+ Left ex ->

985 handleException ex vst

981 handleException ex vst

986 -- The other cases are less interesting, because unless the endpoint is

982 -- The other cases are less interesting, because unless the endpoint is

987 -- in Valid state we're not supposed to do any IO on it

983 -- in Valid state we're not supposed to do any IO on it

@@ -1058,13 +1054,12 @@ withRemoteState :: EndPointPair -- ^ Local and

1058 -> ((MVar (), ValidRemoteEndPointState) -> IO a) -- ^ Case for RemoteEndPointClosing

1054 -> ((MVar (), ValidRemoteEndPointState) -> IO a) -- ^ Case for RemoteEndPointClosing

1059 -> IO a -- ^ Case for RemoteEndPointClosed

1055 -> IO a -- ^ Case for RemoteEndPointClosed

1060 -> IO a

1056 -> IO a

1061−withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do

1057+withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =

1062− modifyRemoteState (ourEndPoint, theirEndPoint)

1058+ modifyRemoteState (ourEndPoint, theirEndPoint)

1063− (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)

1059+ (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)

1064− (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)

1060+ (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)

1065− (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))

1061+ (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))

1066− ((,) RemoteEndPointClosed <$> caseClosed)

1062+ ((,) RemoteEndPointClosed <$> caseClosed)

1067−

1068

1063

1069--------------------------------------------------------------------------------

1064--------------------------------------------------------------------------------

1070-- Incoming requests --

1065-- Incoming requests --

@@ -1094,7 +1089,7 @@ handleConnectionRequest transport sock = handle tryCloseSocket $ do

1094 fail "Invalid endpoint"

1089 fail "Invalid endpoint"

1095 Just ourEndPoint ->

1090 Just ourEndPoint ->

1096 return ourEndPoint

1091 return ourEndPoint

1097− TransportClosed -> do

1092+ TransportClosed ->

1098 fail "Transport closed"

1093 fail "Transport closed"

1099 void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress

1094 void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress

1100 where

1095 where

@@ -1182,7 +1177,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1182 then do

1177 then do

1183 readMessage sock connId

1178 readMessage sock connId

1184 go sock

1179 go sock

1185− else do

1180+ else

1186 case tryToEnum (fromIntegral connId) of

1181 case tryToEnum (fromIntegral connId) of

1187 Just RequestConnectionId -> do

1182 Just RequestConnectionId -> do

1188 recvInt32 sock >>= createNewConnection

1183 recvInt32 sock >>= createNewConnection

@@ -1241,7 +1236,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1241 fail "Local endpoint closed"

1236 fail "Local endpoint closed"

1242 case mmvar of

1237 case mmvar of

1243 Nothing -> do

1238 Nothing -> do

1244− hPutStrLn stderr $ "Warning: Invalid request ID"

1239+ hPutStrLn stderr "Warning: Invalid request ID"

1245 return () -- Invalid request ID. TODO: We just ignore it?

1240 return () -- Invalid request ID. TODO: We just ignore it?

1246 Just mvar ->

1241 Just mvar ->

1247 putMVar mvar response

1242 putMVar mvar response

@@ -1273,8 +1268,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1273 -- We regard a CloseSocket message as an (optimized) way for the

1268 -- We regard a CloseSocket message as an (optimized) way for the

1274 -- remote endpoint to indicate that all its connections to us are

1269 -- remote endpoint to indicate that all its connections to us are

1275 -- now properly closed

1270 -- now properly closed

1276− forM_ (IntSet.elems $ vst ^. remoteIncoming) $ \cid ->

1271+ forM_ (IntSet.elems $ vst ^. remoteIncoming) $

1277− writeChan ourChannel (ConnectionClosed cid)

1272+ writeChan ourChannel . ConnectionClosed

1278 let vst' = remoteIncoming ^= IntSet.empty $ vst

1273 let vst' = remoteIncoming ^= IntSet.empty $ vst

1279 -- Check if we agree that the connection should be closed

1274 -- Check if we agree that the connection should be closed

1280 if vst' ^. remoteOutgoing == 0

1275 if vst' ^. remoteOutgoing == 0