diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,9 @@ +2015-06-15 FacundoDominguez 0.4.2 + +* Update dependencies. +* Fixes in test-suite. +* Bug fixes DP-109, NTTCP-11. + 2014-12-09 Tim Watson 0.4.1 * Update dependencies diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,5 @@ Copyright Well-Typed LLP, 2011-2012 +Copyright Tweag I/O Limited, 2015 All rights reserved. diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -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 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 diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -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 -- diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -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