dee71b3 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start support for TCP parameters

dee71b3Edsko de Vries14 years ago

network-transport/src/Network/Transport/Internal/TCP.hs+4−3

compare with master

unifiedsplit

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

27 )

28import qualified Network.Socket.ByteString as NBS (recv)

29import Control.Concurrent (ThreadId)

30−import Control.Monad (forever)

30+import Control.Monad (forever, when)

31import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)

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

33import Data.ByteString (ByteString)

@@ -54,17 +54,18 @@ import Data.Int (Int32)

54forkServer :: N.HostName -- ^ Host

55 -> N.ServiceName -- ^ Port

56 -> Int -- ^ Backlog (maximum number of queued connections)

57+ -> Bool -- ^ Set ReuseAddr option?

57 -> (SomeException -> IO ()) -- ^ Termination handler

58 -> (SomeException -> IO ()) -- ^ Termination handler

58 -> (N.Socket -> IO ()) -- ^ Request handler

59 -> (N.Socket -> IO ()) -- ^ Request handler

59 -> IO ThreadId

60 -> IO ThreadId

60−forkServer host port backlog terminationHandler requestHandler = do

61+forkServer host port backlog reuseAddr terminationHandler requestHandler = do

61 -- Resolve the specified address. By specification, getAddrInfo will never

62 -- Resolve the specified address. By specification, getAddrInfo will never

62 -- return an empty list (but will throw an exception instead) and will return

63 -- return an empty list (but will throw an exception instead) and will return

63 -- the "best" address first, whatever that means

64 -- the "best" address first, whatever that means

64 addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)

65 addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)

65 bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol)

66 bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol)

66 tryCloseSocket $ \sock -> do

67 tryCloseSocket $ \sock -> do

67− N.setSocketOption sock N.ReuseAddr 1

68+ when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1

68 N.bindSocket sock (N.addrAddress addr)

69 N.bindSocket sock (N.addrAddress addr)

69 N.listen sock backlog

70 N.listen sock backlog

70 mask_ $ forkIOWithUnmask $ \unmask ->

71 mask_ $ forkIOWithUnmask $ \unmask ->

network-transport/src/Network/Transport/TCP.hs+85−40

@@ -12,6 +12,8 @@

12-- compatibility (see "Network.Socket").

13module Network.Transport.TCP ( -- * Main API

14 createTransport

15+ , TCPParameters(..)

16+ , defaultTCPParameters

15 -- * Internals (exposed for unit tests)

17 -- * Internals (exposed for unit tests)

16 , createTransportExposeInternals

18 , createTransportExposeInternals

17 , TransportInternals(..)

19 , TransportInternals(..)

@@ -216,9 +218,10 @@ import Data.Foldable (forM_, mapM_)

216-- ValidRemoteEndPointState).

218-- ValidRemoteEndPointState).

217

219

218data TCPTransport = TCPTransport

220data TCPTransport = TCPTransport

219− { transportHost :: N.HostName

221+ { transportHost :: N.HostName

220− , transportPort :: N.ServiceName

222+ , transportPort :: N.ServiceName

221− , transportState :: MVar TransportState

223+ , transportState :: MVar TransportState

224+ , transportParams :: TCPParameters

222 }

225 }

223

226

224data TransportState =

227data TransportState =

@@ -381,6 +384,19 @@ data ConnectionRequestResponse =

381 | ConnectionRequestCrossed

384 | ConnectionRequestCrossed

382 deriving (Enum, Bounded, Show)

385 deriving (Enum, Bounded, Show)

383

386

387+-- Parameters for setting up the TCP transport

388+data TCPParameters = TCPParameters {

389+ -- | Backlog for 'listen'.

390+ -- Defaults to SOMAXCONN.

391+ tcpBacklog :: Int

392+ -- | Should we set SO_REUSEADDR on the server socket?

393+ -- Defaults to True.

394+ , tcpReuseServerAddr :: Bool

395+ -- | Should we set SO_REUSEADDR on client sockets?

396+ -- Defaults to True.

397+ , tcpReuseClientAddr :: Bool

398+ }

399+

384-- Internal functionality we expose for unit testing

400-- Internal functionality we expose for unit testing

385data TransportInternals = TransportInternals

401data TransportInternals = TransportInternals

386 { -- | The ID of the thread that listens for new incoming connections

402 { -- | The ID of the thread that listens for new incoming connections

@@ -398,42 +414,54 @@ data TransportInternals = TransportInternals

398-- | Create a TCP transport

414-- | Create a TCP transport

399createTransport :: N.HostName

415createTransport :: N.HostName

400 -> N.ServiceName

416 -> N.ServiceName

417+ -> TCPParameters

401 -> IO (Either IOException Transport)

418 -> IO (Either IOException Transport)

402−createTransport host port =

419+createTransport host port params =

403− either Left (Right . fst) <$> createTransportExposeInternals host port

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

404

421

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

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

406createTransportExposeInternals

423createTransportExposeInternals

407 :: N.HostName

424 :: N.HostName

408 -> N.ServiceName

425 -> N.ServiceName

426+ -> TCPParameters

409 -> IO (Either IOException (Transport, TransportInternals))

427 -> IO (Either IOException (Transport, TransportInternals))

410−createTransportExposeInternals host port = do

428+createTransportExposeInternals host port params = do

411 state <- newMVar . TransportValid $ ValidTransportState

429 state <- newMVar . TransportValid $ ValidTransportState

412 { _localEndPoints = Map.empty

430 { _localEndPoints = Map.empty

413 , _nextEndPointId = 0

431 , _nextEndPointId = 0

414 }

432 }

415− let transport = TCPTransport { transportState = state

433+ let transport = TCPTransport { transportState = state

416− , transportHost = host

434+ , transportHost = host

417− , transportPort = port

435+ , transportPort = port

436+ , transportParams = params

418 }

437 }

419− tryIO $ do

438+ tryIO $ bracketOnError (forkServer

420− tid <- forkServer host port N.sOMAXCONN

439+ host

440+ port

441+ (tcpBacklog params)

442+ (tcpReuseServerAddr params)

421− (terminationHandler transport)

443+ (terminationHandler transport)

422− (handleConnectionRequest transport)

444+ (handleConnectionRequest transport))

423− return

445+ killThread

424− ( Transport

446+ (mkTransport transport)

425− { newEndPoint = apiNewEndPoint transport

426− , closeTransport = let evs = [ EndPointClosed

427− , throw $ userError "Transport closed"

428− ] in

429− apiCloseTransport transport (Just tid) evs

430− }

431− , TransportInternals

432− { transportThread = tid

433− , socketBetween = internalSocketBetween transport

434− }

435− )

436 where

447 where

448+ mkTransport :: TCPTransport

449+ -> ThreadId

450+ -> IO (Transport, TransportInternals)

451+ mkTransport transport tid = return

452+ ( Transport

453+ { newEndPoint = apiNewEndPoint transport

454+ , closeTransport = let evs = [ EndPointClosed

455+ , throw $ userError "Transport closed"

456+ ] in

457+ apiCloseTransport transport (Just tid) evs

458+ }

459+ , TransportInternals

460+ { transportThread = tid

461+ , socketBetween = internalSocketBetween transport

462+ }

463+ )

464+

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

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

438 terminationHandler transport ex = do

466 terminationHandler transport ex = do

439 let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))

467 let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))

@@ -441,6 +469,14 @@ createTransportExposeInternals host port = do

441 ]

469 ]

442 apiCloseTransport transport Nothing evs

470 apiCloseTransport transport Nothing evs

443

471

472+-- | Default TCP parameters

473+defaultTCPParameters :: TCPParameters

474+defaultTCPParameters = TCPParameters {

475+ tcpBacklog = N.sOMAXCONN

476+ , tcpReuseServerAddr = True

477+ , tcpReuseClientAddr = True

478+ }

479+

444--------------------------------------------------------------------------------

480--------------------------------------------------------------------------------

445-- API functions --

481-- API functions --

446--------------------------------------------------------------------------------

482--------------------------------------------------------------------------------

@@ -465,7 +501,7 @@ apiNewEndPoint transport = try $ do

465 return EndPoint

501 return EndPoint

466 { receive = readChan (localChannel ourEndPoint)

502 { receive = readChan (localChannel ourEndPoint)

467 , address = localAddress ourEndPoint

503 , address = localAddress ourEndPoint

468− , connect = apiConnect ourEndPoint

504+ , connect = apiConnect (transportParams transport) ourEndPoint

469 , closeEndPoint = let evs = [ EndPointClosed

505 , closeEndPoint = let evs = [ EndPointClosed

470 , throw $ userError "Endpoint closed"

506 , throw $ userError "Endpoint closed"

471 ] in

507 ] in

@@ -480,18 +516,19 @@ apiNewEndPoint transport = try $ do

480 TransportError ResolveMulticastGroupUnsupported "Multicast not supported"

516 TransportError ResolveMulticastGroupUnsupported "Multicast not supported"

481

517

482-- | Connnect to an endpoint

518-- | Connnect to an endpoint

519+apiConnect :: TCPParameters -- ^ Parameters

483−apiConnect :: LocalEndPoint -- ^ Local end point

520+ -> LocalEndPoint -- ^ Local end point

484 -> EndPointAddress -- ^ Remote address

521 -> EndPointAddress -- ^ Remote address

485 -> Reliability -- ^ Reliability (ignored)

522 -> Reliability -- ^ Reliability (ignored)

486 -> ConnectHints -- ^ Hints

523 -> ConnectHints -- ^ Hints

487 -> IO (Either (TransportError ConnectErrorCode) Connection)

524 -> IO (Either (TransportError ConnectErrorCode) Connection)

488−apiConnect ourEndPoint theirAddress _reliability hints =

525+apiConnect params ourEndPoint theirAddress _reliability hints =

489 if localAddress ourEndPoint == theirAddress

526 if localAddress ourEndPoint == theirAddress

490 then connectToSelf ourEndPoint

527 then connectToSelf ourEndPoint

491 else try $ do

528 else try $ do

492 resetIfBroken ourEndPoint theirAddress

529 resetIfBroken ourEndPoint theirAddress

493 (theirEndPoint, connId) <-

530 (theirEndPoint, connId) <-

494− requestConnectionTo ourEndPoint theirAddress hints

531+ requestConnectionTo params ourEndPoint theirAddress hints

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

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

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

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

497 connAlive <- newIORef True

534 connAlive <- newIORef True

@@ -752,11 +789,12 @@ createLocalEndPoint transport = do

752-- additionally block until that is resolved.

789-- additionally block until that is resolved.

753--

790--

754-- May throw a TransportError ConnectErrorCode exception.

791-- May throw a TransportError ConnectErrorCode exception.

755−requestConnectionTo :: LocalEndPoint

792+requestConnectionTo :: TCPParameters

793+ -> LocalEndPoint

756 -> EndPointAddress

794 -> EndPointAddress

757 -> ConnectHints

795 -> ConnectHints

758 -> IO (RemoteEndPoint, ConnectionId)

796 -> IO (RemoteEndPoint, ConnectionId)

759−requestConnectionTo ourEndPoint theirAddress hints = go

797+requestConnectionTo params ourEndPoint theirAddress hints = go

760 where

798 where

761 go = do

799 go = do

762 (theirEndPoint, isNew) <- mapIOException connectFailed $

800 (theirEndPoint, isNew) <- mapIOException connectFailed $

@@ -765,7 +803,7 @@ requestConnectionTo ourEndPoint theirAddress hints = go

765 if isNew

803 if isNew

766 then do

804 then do

767 void . forkEndPointThread ourEndPoint . handle absorbAllExceptions $

805 void . forkEndPointThread ourEndPoint . handle absorbAllExceptions $

768− setupRemoteEndPoint (ourEndPoint, theirEndPoint) hints

806+ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

769 go

807 go

770 else do

808 else do

771 reply <- mapIOException connectFailed $

809 reply <- mapIOException connectFailed $

@@ -779,9 +817,13 @@ requestConnectionTo ourEndPoint theirAddress hints = go

779 return ()

817 return ()

780

818

781-- | Set up a remote endpoint

819-- | Set up a remote endpoint

782−setupRemoteEndPoint :: EndPointPair -> ConnectHints -> IO ()

820+setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()

783−setupRemoteEndPoint (ourEndPoint, theirEndPoint) hints = do

821+setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

784− didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress hints)

822+ didAccept <- bracketOnError (socketToEndPoint

823+ ourAddress

824+ theirAddress

825+ (tcpReuseClientAddr params)

826+ (connectTimeout hints))

785 onError $ \result -> case result of

827 onError $ \result -> case result of

786 Right (sock, ConnectionRequestAccepted) -> do

828 Right (sock, ConnectionRequestAccepted) -> do

787 let vst = ValidRemoteEndPointState

829 let vst = ValidRemoteEndPointState

@@ -861,19 +903,22 @@ resolveInit (ourEndPoint, theirEndPoint) newState =

861-- Maybe throw a TransportError

903-- Maybe throw a TransportError

862socketToEndPoint :: EndPointAddress -- ^ Our address

904socketToEndPoint :: EndPointAddress -- ^ Our address

863 -> EndPointAddress -- ^ Their address

905 -> EndPointAddress -- ^ Their address

864− -> ConnectHints -- ^ Connection hints

906+ -> Bool -- ^ Use SO_REUSEADDR?

907+ -> Maybe Int -- ^ Timeout for connect

865 -> IO (Either (TransportError ConnectErrorCode)

908 -> IO (Either (TransportError ConnectErrorCode)

866 (N.Socket, ConnectionRequestResponse))

909 (N.Socket, ConnectionRequestResponse))

867−socketToEndPoint (EndPointAddress ourAddress) theirAddress hints = try $ do

910+socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =

911+ try $ do

868 (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of

912 (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of

869 Nothing -> throwIO (failed . userError $ "Could not parse")

913 Nothing -> throwIO (failed . userError $ "Could not parse")

870 Just dec -> return dec

914 Just dec -> return dec

871 addr:_ <- mapIOException invalidAddress $

915 addr:_ <- mapIOException invalidAddress $

872 N.getAddrInfo Nothing (Just host) (Just port)

916 N.getAddrInfo Nothing (Just host) (Just port)

873 bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do

917 bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do

918+ when reuseAddr $

874− mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1

919+ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1

875 mapIOException invalidAddress $

920 mapIOException invalidAddress $

876− timeoutMaybe (connectTimeout hints) timeoutError $

921+ timeoutMaybe timeout timeoutError $

877 N.connect sock (N.addrAddress addr)

922 N.connect sock (N.addrAddress addr)

878 response <- mapIOException failed $ do

923 response <- mapIOException failed $ do

879 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])

924 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])

network-transport/tests/TestTCP.hs+24−23

@@ -10,6 +10,7 @@ import Network.Transport.TCP ( createTransport

10 , createTransportExposeInternals

11 , TransportInternals(..)

12 , encodeEndPointAddress

13+ , defaultTCPParameters

13 )

14 )

14import Data.Int (Int32)

15import Data.Int (Int32)

15import Control.Concurrent (threadDelay, killThread)

16import Control.Concurrent (threadDelay, killThread)

@@ -82,7 +83,7 @@ testEarlyDisconnect nextPort = do

82 server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()

83 server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()

83 server serverAddr clientAddr serverDone = do

84 server serverAddr clientAddr serverDone = do

84 tlog "Server"

85 tlog "Server"

85− Right transport <- nextPort >>= createTransport "127.0.0.1"

86+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

86 Right endpoint <- newEndPoint transport

87 Right endpoint <- newEndPoint transport

87 putMVar serverAddr (address endpoint)

88 putMVar serverAddr (address endpoint)

88 theirAddr <- readMVar clientAddr

89 theirAddr <- readMVar clientAddr

@@ -134,7 +135,7 @@ testEarlyDisconnect nextPort = do

134 putMVar clientAddr ourAddress

135 putMVar clientAddr ourAddress

135

136

136 -- Listen for incoming messages

137 -- Listen for incoming messages

137− forkServer "127.0.0.1" clientPort 5 throwIO $ \sock -> do

138+ forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do

138 -- Initial setup

139 -- Initial setup

139 0 <- recvInt32 sock :: IO Int

140 0 <- recvInt32 sock :: IO Int

140 _ <- recvWithLength sock

141 _ <- recvWithLength sock

@@ -160,7 +161,7 @@ testEarlyDisconnect nextPort = do

160 N.sClose sock

161 N.sClose sock

161

162

162 -- Connect to the server

163 -- Connect to the server

163− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr defaultConnectHints

164+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

164

165

165 -- Request a new connection, but don't wait for the response

166 -- Request a new connection, but don't wait for the response

166 let reqId = 0 :: Int32

167 let reqId = 0 :: Int32

@@ -186,7 +187,7 @@ testEarlyCloseSocket nextPort = do

186 server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()

187 server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()

187 server serverAddr clientAddr serverDone = do

188 server serverAddr clientAddr serverDone = do

188 tlog "Server"

189 tlog "Server"

189− Right transport <- nextPort >>= createTransport "127.0.0.1"

190+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

190 Right endpoint <- newEndPoint transport

191 Right endpoint <- newEndPoint transport

191 putMVar serverAddr (address endpoint)

192 putMVar serverAddr (address endpoint)

192 theirAddr <- readMVar clientAddr

193 theirAddr <- readMVar clientAddr

@@ -245,7 +246,7 @@ testEarlyCloseSocket nextPort = do

245 putMVar clientAddr ourAddress

246 putMVar clientAddr ourAddress

246

247

247 -- Listen for incoming messages

248 -- Listen for incoming messages

248− forkServer "127.0.0.1" clientPort 5 throwIO $ \sock -> do

249+ forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do

249 -- Initial setup

250 -- Initial setup

250 0 <- recvInt32 sock :: IO Int

251 0 <- recvInt32 sock :: IO Int

251 _ <- recvWithLength sock

252 _ <- recvWithLength sock

@@ -273,7 +274,7 @@ testEarlyCloseSocket nextPort = do

273 N.sClose sock

274 N.sClose sock

274

275

275 -- Connect to the server

276 -- Connect to the server

276− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr defaultConnectHints

277+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

277

278

278 -- Request a new connection, but don't wait for the response

279 -- Request a new connection, but don't wait for the response

279 let reqId = 0 :: Int32

280 let reqId = 0 :: Int32

@@ -287,14 +288,14 @@ testEarlyCloseSocket nextPort = do

287-- | Test the creation of a transport with an invalid address

288-- | Test the creation of a transport with an invalid address

288testInvalidAddress :: IO N.ServiceName -> IO ()

289testInvalidAddress :: IO N.ServiceName -> IO ()

289testInvalidAddress nextPort = do

290testInvalidAddress nextPort = do

290− Left _ <- nextPort >>= createTransport "invalidHostName"

291+ Left _ <- nextPort >>= \port -> createTransport "invalidHostName" port defaultTCPParameters

291 return ()

292 return ()

292

293

293-- | Test connecting to invalid or non-existing endpoints

294-- | Test connecting to invalid or non-existing endpoints

294testInvalidConnect :: IO N.ServiceName -> IO ()

295testInvalidConnect :: IO N.ServiceName -> IO ()

295testInvalidConnect nextPort = do

296testInvalidConnect nextPort = do

296 port <- nextPort

297 port <- nextPort

297− Right transport <- createTransport "127.0.0.1" port

298+ Right transport <- createTransport "127.0.0.1" port defaultTCPParameters

298 Right endpoint <- newEndPoint transport

299 Right endpoint <- newEndPoint transport

299

300

300 -- Syntax error in the endpoint address

301 -- Syntax error in the endpoint address

@@ -322,7 +323,7 @@ testIgnoreCloseSocket :: IO N.ServiceName -> IO ()

322testIgnoreCloseSocket nextPort = do

323testIgnoreCloseSocket nextPort = do

323 serverAddr <- newEmptyMVar

324 serverAddr <- newEmptyMVar

324 clientDone <- newEmptyMVar

325 clientDone <- newEmptyMVar

325− Right transport <- nextPort >>= createTransport "127.0.0.1"

326+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

326

327

327 forkTry $ server transport serverAddr

328 forkTry $ server transport serverAddr

328 forkTry $ client transport serverAddr clientDone

329 forkTry $ client transport serverAddr clientDone

@@ -360,7 +361,7 @@ testIgnoreCloseSocket nextPort = do

360 let ourAddress = address endpoint

361 let ourAddress = address endpoint

361

362

362 -- Connect to the server

363 -- Connect to the server

363− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr defaultConnectHints

364+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

364

365

365 -- Request a new connection

366 -- Request a new connection

366 tlog "Requesting connection"

367 tlog "Requesting connection"

@@ -405,7 +406,7 @@ testBlockAfterCloseSocket nextPort = do

405 clientAddr <- newEmptyMVar

406 clientAddr <- newEmptyMVar

406 clientDone <- newEmptyMVar

407 clientDone <- newEmptyMVar

407 port <- nextPort

408 port <- nextPort

408− Right transport <- createTransport "127.0.0.1" port

409+ Right transport <- createTransport "127.0.0.1" port defaultTCPParameters

409

410

410 forkTry $ server transport serverAddr clientAddr

411 forkTry $ server transport serverAddr clientAddr

411 forkTry $ client transport serverAddr clientAddr clientDone

412 forkTry $ client transport serverAddr clientAddr clientDone

@@ -440,7 +441,7 @@ testBlockAfterCloseSocket nextPort = do

440 let ourAddress = address endpoint

441 let ourAddress = address endpoint

441

442

442 -- Connect to the server

443 -- Connect to the server

443− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr defaultConnectHints

444+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

444

445

445 -- Request a new connection

446 -- Request a new connection

446 tlog "Requesting connection"

447 tlog "Requesting connection"

@@ -482,7 +483,7 @@ testUnnecessaryConnect nextPort numThreads = do

482 serverAddr <- newEmptyMVar

483 serverAddr <- newEmptyMVar

483

484

484 forkTry $ do

485 forkTry $ do

485− Right transport <- nextPort >>= createTransport "127.0.0.1"

486+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

486 Right endpoint <- newEndPoint transport

487 Right endpoint <- newEndPoint transport

487 putMVar serverAddr (address endpoint)

488 putMVar serverAddr (address endpoint)

488

489

@@ -495,7 +496,7 @@ testUnnecessaryConnect nextPort numThreads = do

495 dones <- replicateM numThreads $ do

496 dones <- replicateM numThreads $ do

496 done <- newEmptyMVar

497 done <- newEmptyMVar

497 forkTry $ do

498 forkTry $ do

498− Right (_, reply) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr defaultConnectHints

499+ Right (_, reply) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

499 case reply of

500 case reply of

500 ConnectionRequestAccepted ->

501 ConnectionRequestAccepted ->

501 putMVar gotAccepted ()

502 putMVar gotAccepted ()

@@ -516,11 +517,11 @@ testUnnecessaryConnect nextPort numThreads = do

516-- | Test that we can create "many" transport instances

517-- | Test that we can create "many" transport instances

517testMany :: IO N.ServiceName -> IO ()

518testMany :: IO N.ServiceName -> IO ()

518testMany nextPort = do

519testMany nextPort = do

519− Right masterTransport <- nextPort >>= createTransport "127.0.0.1"

520+ Right masterTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

520 Right masterEndPoint <- newEndPoint masterTransport

521 Right masterEndPoint <- newEndPoint masterTransport

521

522

522 replicateM_ 10 $ do

523 replicateM_ 10 $ do

523− mTransport <- nextPort >>= createTransport "127.0.0.1"

524+ mTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

524 case mTransport of

525 case mTransport of

525 Left ex -> do

526 Left ex -> do

526 putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex)

527 putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex)

@@ -537,7 +538,7 @@ testMany nextPort = do

537-- | Test what happens when the transport breaks completely

538-- | Test what happens when the transport breaks completely

538testBreakTransport :: IO N.ServiceName -> IO ()

539testBreakTransport :: IO N.ServiceName -> IO ()

539testBreakTransport nextPort = do

540testBreakTransport nextPort = do

540− Right (transport, internals) <- nextPort >>= createTransportExposeInternals "127.0.0.1"

541+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters

541 Right endpoint <- newEndPoint transport

542 Right endpoint <- newEndPoint transport

542

543

543 killThread (transportThread internals) -- Uh oh

544 killThread (transportThread internals) -- Uh oh

@@ -567,7 +568,7 @@ testReconnect nextPort = do

567

568

568 counter <- newMVar (0 :: Int)

569 counter <- newMVar (0 :: Int)

569

570

570− forkServer "127.0.0.1" serverPort 5 throwIO $ \sock -> do

571+ forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do

571 -- Accept the connection

572 -- Accept the connection

572 Right 0 <- tryIO $ (recvInt32 sock :: IO Int)

573 Right 0 <- tryIO $ (recvInt32 sock :: IO Int)

573 Right _ <- tryIO $ recvWithLength sock

574 Right _ <- tryIO $ recvWithLength sock

@@ -596,7 +597,7 @@ testReconnect nextPort = do

596

597

597 -- Client

598 -- Client

598 forkTry $ do

599 forkTry $ do

599− Right transport <- nextPort >>= createTransport "127.0.0.1"

600+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

600 Right endpoint <- newEndPoint transport

601 Right endpoint <- newEndPoint transport

601 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

602 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

602

603

@@ -642,7 +643,7 @@ testUnidirectionalError nextPort = do

642 serverGotPing <- newEmptyMVar

643 serverGotPing <- newEmptyMVar

643

644

644 -- Server

645 -- Server

645− forkServer "127.0.0.1" serverPort 5 throwIO $ \sock -> do

646+ forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do

646 -- We accept connections, but when an exception occurs we don't do

647 -- We accept connections, but when an exception occurs we don't do

647 -- anything (in particular, we don't close the socket). This is important

648 -- anything (in particular, we don't close the socket). This is important

648 -- because when we shutdown one direction of the socket a recv here will

649 -- because when we shutdown one direction of the socket a recv here will

@@ -663,7 +664,7 @@ testUnidirectionalError nextPort = do

663

664

664 -- Client

665 -- Client

665 forkTry $ do

666 forkTry $ do

666− Right (transport, internals) <- nextPort >>= createTransportExposeInternals "127.0.0.1"

667+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters

667 Right endpoint <- newEndPoint transport

668 Right endpoint <- newEndPoint transport

668 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

669 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

669

670

@@ -718,7 +719,7 @@ testUnidirectionalError nextPort = do

718

719

719testInvalidCloseConnection :: IO N.ServiceName -> IO ()

720testInvalidCloseConnection :: IO N.ServiceName -> IO ()

720testInvalidCloseConnection nextPort = do

721testInvalidCloseConnection nextPort = do

721− Right (transport, internals) <- nextPort >>= createTransportExposeInternals "127.0.0.1"

722+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters

722 serverAddr <- newEmptyMVar

723 serverAddr <- newEmptyMVar

723 clientDone <- newEmptyMVar

724 clientDone <- newEmptyMVar

724 serverDone <- newEmptyMVar

725 serverDone <- newEmptyMVar

@@ -773,7 +774,7 @@ main = do

773 , ("InvalidCloseConnection", testInvalidCloseConnection nextPort)

774 , ("InvalidCloseConnection", testInvalidCloseConnection nextPort)

774 ]

775 ]

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

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

776− testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")

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

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

778 case tcpResult of

779 case tcpResult of

779 Left err -> throwIO err

780 Left err -> throwIO err