c071aa9 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Minor cleanup

c071aa9Edsko de Vries14 years ago

network-transport-2/src/Network/Transport.hs+2

compare with master

unifiedsplit

@@ -64,6 +64,8 @@ data Event =

64 deriving Show

66-- | Connection IDs enable receivers to distinguish one connection from another.

67+--

68+-- TODO: Do we want an Int32 at the API level? (Change to Int perhaps?)

67type ConnectionId = Int32

69type ConnectionId = Int32

69-- | Reliability guarantees of a connection.

71-- | Reliability guarantees of a connection.

network-transport-2/src/Network/Transport/TCP.hs+42−32

@@ -46,7 +46,7 @@ import qualified Network.Socket as N ( HostName

46 , connect

47 )

48import Network.Socket.ByteString (sendMany)

49−import Control.Concurrent (forkIO)

49+import Control.Concurrent (forkIO, ThreadId)

50import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)

51import Control.Concurrent.MVar ( MVar

52 , newMVar

@@ -77,6 +77,7 @@ import Data.Map (Map)

77import qualified Data.Map as Map (empty)

78import Data.Lens.Lazy (Lens, lens, intMapLens, mapLens, (^.), (^=), (^+=))

79import Text.Regex.Applicative (RE, few, anySym, sym, many, match)

80+import System.IO (hPutStrLn, stderr)

81-- $design

82-- $design

82--

83--

@@ -275,7 +276,7 @@ createTransport host port = do

275 , transportPort = port

276 , transportPort = port

276 }

277 }

277 tryIO $ do

278 tryIO $ do

278− forkServer host port $ handleConnectionRequest transport

279+ forkServer host port $ void . handleConnectionRequest transport

279 return Transport { newEndPoint = apiNewEndPoint transport }

280 return Transport { newEndPoint = apiNewEndPoint transport }

280

281

281--------------------------------------------------------------------------------

282--------------------------------------------------------------------------------

@@ -340,14 +341,12 @@ apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (E

340apiSend theirEndPoint connId connAlive payload = do

341apiSend theirEndPoint connId connAlive payload = do

341 withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) -> do

342 withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) -> do

342 alive <- readIORef connAlive

343 alive <- readIORef connAlive

343− if alive

344+ runErrorT $ if alive

344− then do

345+ then

346+ failWithIO (FailedWith SendFailed . show) $

345− result <- tryIO $ sendOn remoteConn (encodeInt32 connId : prependLength payload)

347+ sendOn remoteConn (encodeInt32 connId : prependLength payload)

346− case result of

347− Left err -> return . Left $ FailedWith SendFailed (show err)

348− Right _ -> return . Right $ ()

349− else do

348+ else

350− return . Left $ FailedWith SendConnectionClosed "Connection closed"

349+ throwError $ FailedWith SendConnectionClosed "Connection closed"

351

350

352-- | Special case of 'apiConnect': connect an endpoint to itself

351-- | Special case of 'apiConnect': connect an endpoint to itself

353connectToSelf :: LocalEndPoint -> IO (Either (FailedWith ConnectErrorCode) Connection)

352connectToSelf :: LocalEndPoint -> IO (Either (FailedWith ConnectErrorCode) Connection)

@@ -449,12 +448,6 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go

449 -- matter.

448 -- matter.

450 case endPointStateSnapshot of

449 case endPointStateSnapshot of

451 RemoteEndPointInvalid err -> do

450 RemoteEndPointInvalid err -> do

452− -- We remove the endpoint from the state again because the next call

453− -- to 'connect' might give a different result. (We definitely don't

454− -- want to do this while holding the endpoint lock, because holding

455− -- multiple locks introduces all kinds of trouble.)

456− modifyMVar ourState $ \st ->

457− return (connectionTo theirAddress ^= Nothing $ st, Left err)

458 return . Left $ err

451 return . Left $ err

459

452

460 RemoteEndPointClosing _ resolved -> do

453 RemoteEndPointClosing _ resolved -> do

@@ -475,8 +468,8 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go

475 -- endpoint replies (but note that we don't hold any locks at this

468 -- endpoint replies (but note that we don't hold any locks at this

476 -- point)

469 -- point)

477 mcid <- tryIO $ do

470 mcid <- tryIO $ do

478− reply <- sendRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

471+ reply <- doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

479− decodeInt32 . BS.concat <$> takeMVar reply

472+ return (decodeInt32 . BS.concat $ reply)

480

473

481 -- On a failure we decrement the refcount again and return an error.

474 -- On a failure we decrement the refcount again and return an error.

482 -- The only state the remote endpoint can be in at this point is

475 -- The only state the remote endpoint can be in at this point is

@@ -499,7 +492,9 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go

499-- RemoteEndPointInvalid.

492-- RemoteEndPointInvalid.

500connectToRemoteEndPoint :: EndPointPair -> IO ()

493connectToRemoteEndPoint :: EndPointPair -> IO ()

501connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do

494connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do

495+ let ourState = localState ourEndPoint

502− let theirState = remoteState theirEndPoint

496+ theirState = remoteState theirEndPoint

497+ theirAddress = remoteAddress theirEndPoint

503 result <- runErrorT go

498 result <- runErrorT go

504 case result of

499 case result of

505 Right (ConnectionRequestAccepted, sock) -> do

500 Right (ConnectionRequestAccepted, sock) -> do

@@ -510,14 +505,22 @@ connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do

510 putMVar theirState (RemoteEndPointValid remoteConn)

505 putMVar theirState (RemoteEndPointValid remoteConn)

511 void . forkIO $ handleIncomingMessages (ourEndPoint, theirEndPoint)

506 void . forkIO $ handleIncomingMessages (ourEndPoint, theirEndPoint)

512 Right (ConnectionRequestEndPointInvalid, sock) -> do

507 Right (ConnectionRequestEndPointInvalid, sock) -> do

508+ -- We remove the endpoint from our local state again because the next

509+ -- call to 'connect' might give a different result. Threads that were

510+ -- waiting on the result of this call to connect will get the

511+ -- RemoteEndPointInvalid; subsequent threads will initiate a new

512+ -- connection requests.

513+ modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)

513 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))

514 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))

514 N.sClose sock

515 N.sClose sock

515 Right (ConnectionRequestCrossed, sock) -> do

516 Right (ConnectionRequestCrossed, sock) -> do

516 N.sClose sock

517 N.sClose sock

517 Left err -> do

518 Left err -> do

519+ -- See comment above

520+ modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)

521+ putMVar theirState (RemoteEndPointInvalid err)

518 -- TODO: this isn't quite right; socket *might* have been opened here,

522 -- TODO: this isn't quite right; socket *might* have been opened here,

519 -- in which case we should close it

523 -- in which case we should close it

520− putMVar theirState (RemoteEndPointInvalid err)

521 where

524 where

522 go :: ErrorT (FailedWith ConnectErrorCode) IO (ConnectionRequestResponse, N.Socket)

525 go :: ErrorT (FailedWith ConnectErrorCode) IO (ConnectionRequestResponse, N.Socket)

523 go = do

526 go = do

@@ -553,20 +556,19 @@ decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack

553 endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)

556 endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)

554 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))

557 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))

555

558

556−-- | Send a remote request and return the MVar that will contain the response

559+-- | Do a (blocking) remote request

557--

560--

558-- RELY: Remote endpoint must be in valid state.

561-- RELY: Remote endpoint must be in valid state.

559-- GUARANTEE: Will not change the state of the remote endpoint.

562-- GUARANTEE: Will not change the state of the remote endpoint.

560−sendRemoteRequest :: EndPointPair -> ControlHeader -> IO (MVar [ByteString])

563+doRemoteRequest :: EndPointPair -> ControlHeader -> IO ([ByteString])

561−sendRemoteRequest (ourEndPoint, theirEndPoint) header = do

564+doRemoteRequest (ourEndPoint, theirEndPoint) header = do

562 reply <- newEmptyMVar

565 reply <- newEmptyMVar

563 reqId <- modifyMVar (localState ourEndPoint) $ \st -> do

566 reqId <- modifyMVar (localState ourEndPoint) $ \st -> do

564 let reqId = st ^. nextCtrlRequestId

567 let reqId = st ^. nextCtrlRequestId

565 return ((nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ st, reqId)

568 return ((nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ st, reqId)

566 withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) ->

569 withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) ->

567 sendOn remoteConn [encodeInt32 header, encodeInt32 reqId]

570 sendOn remoteConn [encodeInt32 header, encodeInt32 reqId]

568− return reply

571+ takeMVar reply

569−

570

572

571-- | Decrease the remote refcount of a remote endpoint. When the refcount

573-- | Decrease the remote refcount of a remote endpoint. When the refcount

572-- reaches zero, initiate socket closure.

574-- reaches zero, initiate socket closure.

@@ -591,7 +593,9 @@ decreaseRemoteRefCount theirEndPoint = do

591

593

592-- | Handle a connection request (that is, a remote endpoint that is trying to

594-- | Handle a connection request (that is, a remote endpoint that is trying to

593-- establish a TCP connection with us)

595-- establish a TCP connection with us)

596+--

597+-- Returns the thread ID of the thread that is handling incoming messages.

594−handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()

598+handleConnectionRequest :: TCPTransport -> N.Socket -> IO (Maybe ThreadId)

595handleConnectionRequest transport sock = do

599handleConnectionRequest transport sock = do

596 request <- tryIO $ do

600 request <- tryIO $ do

597 ourEndPointId <- recvInt32 sock

601 ourEndPointId <- recvInt32 sock

@@ -602,6 +606,7 @@ handleConnectionRequest transport sock = do

602 Right (_, Nothing) -> do

606 Right (_, Nothing) -> do

603 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]

607 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]

604 N.sClose sock

608 N.sClose sock

609+ return Nothing

605 Right (theirAddress, Just ourEndPoint) -> do

610 Right (theirAddress, Just ourEndPoint) -> do

606 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->

611 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->

607 case st ^. connectionTo theirAddress of

612 case st ^. connectionTo theirAddress of

@@ -617,23 +622,24 @@ handleConnectionRequest transport sock = do

617 then do

622 then do

618 sendMany sock [encodeInt32 ConnectionRequestCrossed]

623 sendMany sock [encodeInt32 ConnectionRequestCrossed]

619 N.sClose sock

624 N.sClose sock

625+ return Nothing

620 else do

626 else do

621 let remoteConn = RemoteConnection { remoteSocket = sock

627 let remoteConn = RemoteConnection { remoteSocket = sock

622 , _remoteRefCount = 0

628 , _remoteRefCount = 0

623 , sendOn = sendMany sock

629 , sendOn = sendMany sock

624 }

630 }

625 putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)

631 putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)

626− forkIO $ handleIncomingMessages (ourEndPoint, theirEndPoint)

627 sendMany sock [encodeInt32 ConnectionRequestAccepted]

632 sendMany sock [encodeInt32 ConnectionRequestAccepted]

633+ tid <- forkIO $ handleIncomingMessages (ourEndPoint, theirEndPoint)

634+ return (Just tid)

628− Left _ ->

635+ Left _ -> do

629 -- Invalid request

636 -- Invalid request

630 N.sClose sock

637 N.sClose sock

638+ return Nothing

631

639

632-- | Handle requests from a remote endpoint.

640-- | Handle requests from a remote endpoint.

633--

641--

634-- Returns only if the remote party closes the socket or if an error occurs.

642-- Returns only if the remote party closes the socket or if an error occurs.

635−-- The return value is the list of connections that are still open (normally,

636−-- this should be an empty list).

637--

643--

638-- RELY: The remote endpoint must be in RemoteEndPointValid or

644-- RELY: The remote endpoint must be in RemoteEndPointValid or

639-- RemoteEndPointClosing state. If the latter, then the 'resolved' MVar

645-- RemoteEndPointClosing state. If the latter, then the 'resolved' MVar

@@ -684,7 +690,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

684 go sock openConnections

690 go sock openConnections

685 Nothing ->

691 Nothing ->

686 -- Invalid control request, exit

692 -- Invalid control request, exit

687− putStrLn "Warning: invalid control request"

693+ hPutStrLn stderr "Warning: invalid control request"

688

694

689 -- Create a new connection

695 -- Create a new connection

690 createNewConnection :: IORef IntSet -> ControlRequestId -> IO ()

696 createNewConnection :: IORef IntSet -> ControlRequestId -> IO ()

@@ -722,7 +728,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

722 mmvar <- modifyMVar ourState $ \st ->

728 mmvar <- modifyMVar ourState $ \st ->

723 return (pendingCtrlRequestsAt reqId ^= Nothing $ st, st ^. pendingCtrlRequestsAt reqId)

729 return (pendingCtrlRequestsAt reqId ^= Nothing $ st, st ^. pendingCtrlRequestsAt reqId)

724 case mmvar of

730 case mmvar of

725− Nothing ->

731+ Nothing -> do

732+ hPutStrLn stderr $ "Warning: Invalid request ID"

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

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

727 Just mvar ->

734 Just mvar ->

728 putMVar mvar response

735 putMVar mvar response

@@ -744,6 +751,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

744 N.sClose sock

751 N.sClose sock

745 return (RemoteEndPointClosed, True)

752 return (RemoteEndPointClosed, True)

746 RemoteEndPointClosing _ resolved -> do

753 RemoteEndPointClosing _ resolved -> do

754+ -- If the socket is already in closing state, then we have already

755+ -- sent a CloseSocket request; hence, we don't need to do it again

756+ -- at this point

747 putMVar resolved ()

757 putMVar resolved ()

748 N.sClose sock

758 N.sClose sock

749 return (RemoteEndPointClosed, True)

759 return (RemoteEndPointClosed, True)