network-transport-tcp 0.4.1 → 0.4.2
raw patch · 5 files changed
+84/−54 lines, 5 filesnew-uploaderPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- ChangeLog +6/−0
- LICENSE +1/−0
- network-transport-tcp.cabal +6/−6
- src/Network/Transport/TCP.hs +49/−38
- tests/TestTCP.hs +22/−10
ChangeLog view
@@ -1,3 +1,9 @@+2015-06-15 FacundoDominguez <facundo.dominguez@tweag.io> 0.4.2++* Update dependencies.+* Fixes in test-suite.+* Bug fixes DP-109, NTTCP-11.+ 2014-12-09 Tim Watson <watson.timothy@gmail.com> 0.4.1 * Update dependencies
LICENSE view
@@ -1,4 +1,5 @@ Copyright Well-Typed LLP, 2011-2012+Copyright Tweag I/O Limited, 2015 All rights reserved.
network-transport-tcp.cabal view
@@ -1,18 +1,18 @@ Name: network-transport-tcp-Version: 0.4.1+Version: 0.4.2 Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 License-file: LICENSE-Copyright: Well-Typed LLP+Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer: edsko@well-typed.com, duncan@well-typed.com, watson.timothy@gmail.com+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io> Stability: experimental Homepage: http://haskell-distributed.github.com Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTTCP Synopsis: TCP instantiation of Network.Transport Description: TCP instantiation of Network.Transport -Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2+Tested-With: GHC==7.4.2 GHC==7.6.3 GHC==7.8.4 GHC==7.10.1 Category: Network extra-source-files: ChangeLog @@ -53,7 +53,7 @@ network-transport-tcp >= 0.3 && < 0.5 ghc-options: -threaded -rtsopts -with-rtsopts=-N HS-Source-Dirs: tests- Extensions: CPP,+ default-extensions: CPP, OverloadedStrings default-language: Haskell2010 If flag(use-mock-network)@@ -83,7 +83,7 @@ Buildable: False ghc-options: -threaded -Wall -fno-warn-orphans HS-Source-Dirs: tests- Extensions: TypeSynonymInstances+ default-extensions: TypeSynonymInstances FlexibleInstances OverlappingInstances OverloadedStrings
src/Network/Transport/TCP.hs view
@@ -101,7 +101,7 @@ ) import Control.Category ((>>>)) import Control.Applicative ((<$>))-import Control.Monad (when, unless, join)+import Control.Monad (when, unless, join, (<=<)) import Control.Exception ( IOException , SomeException@@ -112,9 +112,12 @@ , try , bracketOnError , fromException+ , finally , catch+ , bracket+ , mask_ )-import Data.IORef (IORef, newIORef, writeIORef, readIORef)+import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef) import Data.ByteString (ByteString) import qualified Data.ByteString as BS (concat) import qualified Data.ByteString.Char8 as BSC (pack, unpack)@@ -132,6 +135,7 @@ ) import Data.Map (Map) import qualified Data.Map as Map (empty)+import Data.Traversable (traverse) import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_)@@ -603,26 +607,25 @@ -- | Close a connection apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO () apiClose (ourEndPoint, theirEndPoint) connId connAlive =- void . tryIO . asyncWhenCancelled return $ do- mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointValid vst -> do- alive <- readIORef connAlive- if alive- then do- writeIORef connAlive False- act <- schedule theirEndPoint $- sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]- return ( RemoteEndPointValid- . (remoteOutgoing ^: (\x -> x - 1))- $ vst- , Just act- )- else- return (RemoteEndPointValid vst, Nothing)- _ ->- return (st, Nothing)- forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)- closeIfUnused (ourEndPoint, theirEndPoint)+ void . tryIO . asyncWhenCancelled return $ finally+ (withScheduledAction ourEndPoint $ \sched -> do+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointValid vst -> do+ alive <- readIORef connAlive+ if alive+ then do+ writeIORef connAlive False+ sched theirEndPoint $+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]+ return ( RemoteEndPointValid+ . (remoteOutgoing ^: (\x -> x - 1))+ $ vst+ )+ else+ return (RemoteEndPointValid vst)+ _ ->+ return st)+ (closeIfUnused (ourEndPoint, theirEndPoint)) -- | Send data across a connection@@ -633,8 +636,8 @@ -> IO (Either (TransportError SendErrorCode) ()) apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = -- We don't need the overhead of asyncWhenCancelled here- try . mapIOException sendFailed $ do- act <- withMVar (remoteState theirEndPoint) $ \st -> case st of+ try . mapIOException sendFailed $ withScheduledAction ourEndPoint $ \sched -> do+ withMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "apiSend" RemoteEndPointInit _ _ _ ->@@ -642,7 +645,7 @@ RemoteEndPointValid vst -> do alive <- readIORef connAlive if alive- then schedule theirEndPoint $+ then sched theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload) else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointClosing _ _ -> do@@ -660,7 +663,6 @@ if alive then throwIO $ TransportError SendFailed (show err) else throwIO $ TransportError SendClosed "Connection closed"- runScheduledAction (ourEndPoint, theirEndPoint) act where sendFailed = TransportError SendFailed . show @@ -686,33 +688,32 @@ where -- Close the remote socket and return the set of all incoming connections tryCloseRemoteSocket :: RemoteEndPoint -> IO ()- tryCloseRemoteSocket theirEndPoint = do+ tryCloseRemoteSocket theirEndPoint = withScheduledAction ourEndPoint $ \sched -> do -- We make an attempt to close the connection nicely -- (by sending a CloseSocket first) let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"- mAct <- modifyMVar (remoteState theirEndPoint) $ \st ->+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInvalid _ ->- return (st, Nothing)+ return st RemoteEndPointInit resolved _ _ -> do putMVar resolved ()- return (closed, Nothing)+ return closed RemoteEndPointValid vst -> do- act <- schedule theirEndPoint $ do+ sched theirEndPoint $ do tryIO $ sendOn vst [ encodeInt32 CloseSocket , encodeInt32 (vst ^. remoteMaxIncoming) ] tryCloseSocket (remoteSocket vst)- return (closed, Just act)+ return closed RemoteEndPointClosing resolved vst -> do putMVar resolved ()- act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)- return (closed, Just act)+ sched theirEndPoint $ tryCloseSocket (remoteSocket vst)+ return closed RemoteEndPointClosed ->- return (st, Nothing)+ return st RemoteEndPointFailed err ->- return (RemoteEndPointFailed err, Nothing)- forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)+ return (RemoteEndPointFailed err) --------------------------------------------------------------------------------@@ -1200,7 +1201,8 @@ LocalEndPointValid _ -> do alive <- readIORef connAlive if alive- then writeChan ourChan (Received connId msg)+ then seq (foldr seq () msg)+ writeChan ourChan (Received connId msg) else throwIO $ TransportError SendClosed "Connection closed" LocalEndPointClosed -> throwIO $ TransportError SendFailed "Endpoint closed"@@ -1464,6 +1466,15 @@ err = TransportError code (show ex) writeChan (localChannel ourEndPoint) $ ErrorEvent err return (RemoteEndPointFailed ex)++-- | Use 'schedule' action 'runScheduled' action in a safe way, it's assumed that+-- callback is used only once, otherwise guarantees of runScheduledAction are not+-- respected.+withScheduledAction :: LocalEndPoint -> ((RemoteEndPoint -> IO a -> IO ()) -> IO ()) -> IO ()+withScheduledAction ourEndPoint f =+ bracket (newIORef Nothing)+ (traverse (\(tp, a) -> runScheduledAction (ourEndPoint, tp) a) <=< readIORef)+ (\ref -> f (\rp g -> mask_ $ schedule rp g >>= \x -> writeIORef ref (Just (rp,x)) )) -------------------------------------------------------------------------------- -- "Stateless" (MVar free) functions --
tests/TestTCP.hs view
@@ -590,24 +590,25 @@ -- Accept the connection Right 0 <- tryIO $ (recvInt32 sock :: IO Int) Right _ <- tryIO $ recvWithLength sock- Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] -- The first time we close the socket before accepting the logical connection count <- modifyMVar counter $ \i -> return (i + 1, i) when (count > 0) $ do+ Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] -- Client requests a logical connection- Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)- connId <- recvInt32 sock :: IO LightweightConnectionId- return ()- when (count > 1) $ do- -- Client sends a message- Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)- True <- return $ connId == connId'- Right ["ping"] <- tryIO $ recvWithLength sock- putMVar serverDone ()+ Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)+ connId <- recvInt32 sock :: IO LightweightConnectionId+ return () + when (count > 2) $ do+ -- Client sends a message+ Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)+ True <- return $ connId == connId'+ Right ["ping"] <- tryIO $ recvWithLength sock+ putMVar serverDone ()+ Right () <- tryIO $ N.sClose sock return () @@ -633,6 +634,17 @@ Just (Left (TransportError ConnectFailed _)) -> return () Just (Left err) -> throwIO err Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success"++ resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints+ case resultConnect of+ Nothing -> return ()+ Just (Left (TransportError ConnectFailed _)) -> return ()+ Just (Left err) -> throwIO err+ Just (Right c) -> do+ ev <- send c ["foo"]+ case ev of+ Left _ -> return ()+ Right _ -> throwIO $ userError "testConnect: unexpected send success" -- The third attempt succeeds Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints