806436c · distributed-process

git.iko.soy/distributed-process

Reduce overhead in apiSend

Since previously we caught *all* exceptions in modifyRemoteState we need to distinguish between various kinds of exceptions in apiSend manually. Now we only catch IO exceptions, so we can throw TransportErrors normally.

commit806436c5477eb391c78eee924627fba51780b2f9 authorEdsko de Vries <edsko@edsko.net> dateMon 28 May 2012, 11:29 (14 years ago) parentb901caa

2 files changed+49−28

compare with master

unifiedsplit

network-transport/src/Network/Transport/TCP.hs+47−25 network-transport/tests/TestTransport.hs+2−3

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

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

7272 )

7373import Control.Category ((>>>))

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

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

75+import Control.Monad (when, unless)

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

7777import Control.Exception ( IOException

7878 , SomeException

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

250250 , _nextRemoteId :: !Int

251251 }

252252

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+--

253290-- Invariants for dealing with remote endpoints:

254291--

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

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

269306-- both signal on 'resolved'.

270307--

271308-- 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

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

274311-- once.

275312--

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

448485 }

449486

450487-- | 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.

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

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

458490 modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

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

472504 closeIfUnused (ourEndPoint, theirEndPoint)

473505

474506-- | 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.

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

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

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

481510 -> [ByteString] -- ^ Payload

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

483512apiSend (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 $

489514 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

490515 { caseInvalid = \_ ->

491516 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

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

494519 , caseValid = \vst -> do

495520 alive <- readIORef connAlive

496521 if alive

497− then do

498− sendOn vst (encodeInt32 connId : prependLength payload)

499− return . Right $ ()

500− else

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

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

523+ else throwIO $ TransportError SendClosed "Connection closed"

502524 , caseClosing = \_ _ -> do

503525 alive <- readIORef connAlive

504526 if alive

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

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

527+ then throwIO $ TransportError SendFailed "Connection lost"

528+ else throwIO $ TransportError SendClosed "Connection closed"

507529 , caseClosed = do

508530 alive <- readIORef connAlive

509531 if alive

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

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

532+ then throwIO $ TransportError SendFailed "Connection lost"

533+ else throwIO $ TransportError SendClosed "Connection closed"

512534 }

513535 where

514536 sendFailed = TransportError SendFailed . show

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

704726 case decodeInt32 . BS.concat $ reply of

705727 Nothing ->

706728 throwIO (connectFailed $ userError "Invalid integer")

707− Just cid -> do

729+ Just cid ->

708730 return (theirEndPoint, cid)

709731

710732 connectFailed = TransportError ConnectFailed . show

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

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

650650 send conn ["pong"]

651651

652652 -- 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

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

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

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

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

657656

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