806436c · distributed-process

git.iko.soy/distributed-process

on master help sign in

Reduce overhead in apiSend

806436cEdsko de Vries14 years ago

network-transport/src/Network/Transport/TCP.hs+47−25

compare with master

unifiedsplit

@@ -72,7 +72,7 @@ import Control.Concurrent.MVar ( MVar

72 )

73import Control.Category ((>>>))

74import Control.Applicative ((<$>))

75−import Control.Monad (when, unless, join, liftM)

75+import Control.Monad (when, unless)

76import Control.Monad.Error (ErrorT(..), runErrorT)

77import Control.Exception ( IOException

78 , SomeException

@@ -250,6 +250,43 @@ data ValidLocalEndPointState = ValidLocalEndPointState

250 , _nextRemoteId :: !Int

251 }

252

253+-- REMOTE ENDPOINTS

254+--

255+-- Remote endpoints (basically, TCP connections) have the following lifecycle:

256+--

257+-- Init ----+---> Invalid

258+-- |

259+-- | /----------------\

260+-- | | |

261+-- | v |

262+-- \---> Valid ----+---> Closing

263+-- | |

264+-- | v

265+-- \---> Closed

266+--

267+-- Init: There are two places where we create new remote endpoints: in

268+-- requestConnectionTo (in response to an API 'connect' call) and in

269+-- handleConnectionRequest (when a remote node tries to connect to us).

270+-- 'Init' carries an MVar () 'resolved' which concurrent threads can use to

271+-- wait for the remote endpoint to finish initialization.

272+--

273+-- Invalid: We put the remote endpoint in invalid state only during

274+-- requestConnectionTo when we fail to connect.

275+--

276+-- Valid: This is the "normal" state for a working remote endpoint.

277+--

278+-- Closing: When we detect that a remote endpoint is no longer used, we send a

279+-- CloseSocket request across the connection and put the remote endpoint in

280+-- closing state. As with Init, 'Closing' carries an MVar () 'resolved' which

281+-- concurrent threads can use to wait for the remote endpoint to either be

282+-- closed fully (if the communication parnet responds with another

283+-- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies

284+-- the request.

285+--

286+-- Closed: The endpoint is put in Closed state after a successful garbage

287+-- collection, when the endpoint (or the whole transport) is closed manually,

288+-- or when an IO exception occurs during communication.

289+--

253-- Invariants for dealing with remote endpoints:

290-- Invariants for dealing with remote endpoints:

254--

291--

255-- INV-SEND: Whenever we send data the remote endpoint must be locked (to avoid

292-- INV-SEND: Whenever we send data the remote endpoint must be locked (to avoid

@@ -269,7 +306,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState

269-- both signal on 'resolved'.

306-- both signal on 'resolved'.

270--

307--

271-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we

308-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we

272−-- first put the remote endpoint in Closing or Closed state, and then send a

309+-- first put the remote endpoint in Closed state, and then send a

273-- EventConnectionLost event. This guarantees that we only send this event

310-- EventConnectionLost event. This guarantees that we only send this event

274-- once.

311-- once.

275--

312--

@@ -448,11 +485,6 @@ apiConnect ourEndPoint theirAddress _reliability _hints =

448 }

485 }

449

486

450-- | Close a connection

487-- | Close a connection

451−--

452−-- RELY: The endpoint must not be invalid

453−-- GUARANTEE: If the connection is alive on entry then the remote endpoint will

454−-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the

455−-- state of the remote endpoint will not be changed.

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

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

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

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

458 modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

490 modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

@@ -472,20 +504,13 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do

472 closeIfUnused (ourEndPoint, theirEndPoint)

504 closeIfUnused (ourEndPoint, theirEndPoint)

473

505

474-- | Send data across a connection

506-- | Send data across a connection

475−--

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

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

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

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

479 -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)

508 -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)

480 -> IORef Bool -- ^ Is the connection still alive?

509 -> IORef Bool -- ^ Is the connection still alive?

481 -> [ByteString] -- ^ Payload

510 -> [ByteString] -- ^ Payload

482 -> IO (Either (TransportError SendErrorCode) ())

511 -> IO (Either (TransportError SendErrorCode) ())

483apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

512apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

484− -- The 'join' joins the inner exception (which we explicitly return, for

485− -- instance if the connection is closed) with the outer exception (which is

486− -- returned by 'try' when an exception is thrown by 'sendOn', and handled

487− -- by 'withRemoteState')

488− liftM join . try . mapIOException sendFailed $

513+ try . mapIOException sendFailed $

489 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

514 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

490 { caseInvalid = \_ ->

515 { caseInvalid = \_ ->

491 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

516 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

@@ -494,21 +519,18 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

494 , caseValid = \vst -> do

519 , caseValid = \vst -> do

495 alive <- readIORef connAlive

520 alive <- readIORef connAlive

496 if alive

521 if alive

497− then do

498− sendOn vst (encodeInt32 connId : prependLength payload)

522+ then sendOn vst (encodeInt32 connId : prependLength payload)

499− return . Right $ ()

500− else

501− return . Left $ TransportError SendClosed "Connection closed"

523+ else throwIO $ TransportError SendClosed "Connection closed"

502 , caseClosing = \_ _ -> do

524 , caseClosing = \_ _ -> do

503 alive <- readIORef connAlive

525 alive <- readIORef connAlive

504 if alive

526 if alive

505− then return . Left $ TransportError SendFailed "Connection lost"

527+ then throwIO $ TransportError SendFailed "Connection lost"

506− else return . Left $ TransportError SendClosed "Connection closed"

528+ else throwIO $ TransportError SendClosed "Connection closed"

507 , caseClosed = do

529 , caseClosed = do

508 alive <- readIORef connAlive

530 alive <- readIORef connAlive

509 if alive

531 if alive

510− then return . Left $ TransportError SendFailed "Connection lost"

532+ then throwIO $ TransportError SendFailed "Connection lost"

511− else return . Left $ TransportError SendClosed "Connection closed"

533+ else throwIO $ TransportError SendClosed "Connection closed"

512 }

534 }

513 where

535 where

514 sendFailed = TransportError SendFailed . show

536 sendFailed = TransportError SendFailed . show

@@ -704,7 +726,7 @@ requestConnectionTo ourEndPoint theirAddress = go

704 case decodeInt32 . BS.concat $ reply of

726 case decodeInt32 . BS.concat $ reply of

705 Nothing ->

727 Nothing ->

706 throwIO (connectFailed $ userError "Invalid integer")

728 throwIO (connectFailed $ userError "Invalid integer")

707− Just cid -> do

729+ Just cid ->

708 return (theirEndPoint, cid)

730 return (theirEndPoint, cid)

709

731

710 connectFailed = TransportError ConnectFailed . show

732 connectFailed = TransportError ConnectFailed . show

network-transport/tests/TestTransport.hs+2−3

@@ -650,9 +650,8 @@ testCloseTransport newTransport = do

650 send conn ["pong"]

651

652 -- Client now closes down its transport. We should receive connection closed messages

653− -- TODO: this assumes a certain ordering on the messages we receive; that's not guaranteed

654− ConnectionClosed cid1' <- receive endpoint ; True <- return $ cid1' == cid1

653+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid' == cid1 || cid' == cid2

655− ConnectionClosed cid2'' <- receive endpoint ; True <- return $ cid2'' == cid2

654+ ConnectionClosed cid'' <- receive endpoint ; True <- return $ (cid'' == cid1 || cid'' == cid2) && cid'' /= cid'

656 ErrorEvent (TransportError (EventConnectionLost addr'' []) _) <- receive endpoint ; True <- return $ addr'' == theirAddr2

655 ErrorEvent (TransportError (EventConnectionLost addr'' []) _) <- receive endpoint ; True <- return $ addr'' == theirAddr2

657

656

658 -- An attempt to send to the endpoint should now fail

657 -- An attempt to send to the endpoint should now fail