diff --git a/ChangeLog b/ChangeLog new file mode 100644 --- /dev/null +++ b/ChangeLog @@ -0,0 +1,30 @@ +2014-05-30 Tim Watson 0.4.0.0 + +* Fix build for GHC 7.4 - thanks mboes! +* Allow transformers above v5 +* Bump binary version to include 0.7.* +* Binary instance for 'Reliability' - thanks mboes! +* Hashable instance for 'EndPointAddress' + +2012-11-22 Edsko de Vries 0.3.0.1 + +* Relax bounds on Binary + +2012-10-03 Edsko de Vries 0.3.0 + +* Clarify disconnection +* Require that 'connect' be "as asynchronous as possible" +* Added strictness annotations + +2012-07-16 Edsko de Vries 0.2.0.2 + +* Base 4.6 compatible test suites +* Relax package constraints for bytestring + +2012-07-16 Edsko de Vries 0.2.0.1 + +* Hide catch only for base < 4.6 + +2012-07-07 Edsko de Vries 0.2.0 + +* Initial release. diff --git a/network-transport.cabal b/network-transport.cabal --- a/network-transport.cabal +++ b/network-transport.cabal @@ -1,15 +1,15 @@ Name: network-transport -Version: 0.3.0.1 +Version: 0.4.0.0 Cabal-Version: >=1.6 Build-Type: Simple License: BSD3 License-File: LICENSE Copyright: Well-Typed LLP Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: edsko@well-typed.com, duncan@well-typed.com +Maintainer: edsko@well-typed.com, duncan@well-typed.com, watson.timothy@gmail.com Stability: experimental -Homepage: http://github.com/haskell-distributed/distributed-process -Bug-Reports: mailto:edsko@well-typed.com +Homepage: http://haskell-distributed.github.com +Bug-Reports: https://cloud-haskell.atlassian.net/browse/NT Synopsis: Network abstraction layer Description: "Network.Transport" is a Network Abstraction Layer which provides the following high-level concepts: @@ -56,18 +56,21 @@ probably also want to install at least one transport implementation (network-transport-*). Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 -Category: Network +Category: Network +extra-source-files: ChangeLog Source-Repository head Type: git - Location: https://github.com/haskell-distributed/distributed-process - SubDir: network-transport + Location: https://github.com/haskell-distributed/network-transport Library Build-Depends: base >= 4.3 && < 5, - binary >= 0.5 && < 0.7, + binary >= 0.5 && < 0.8, bytestring >= 0.9 && < 0.11, - transformers >= 0.2 && < 0.4 + hashable >= 1.2.0.5 && < 1.3, + transformers >= 0.2 && < 0.5 + if impl(ghc < 7.6) + Build-Depends: ghc-prim >= 0.2 && < 0.4 Exposed-Modules: Network.Transport, Network.Transport.Util Network.Transport.Internal diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs --- a/src/Network/Transport.hs +++ b/src/Network/Transport.hs @@ -1,5 +1,5 @@ --- | Network Transport -module Network.Transport +-- | Network Transport +module Network.Transport ( -- * Types Transport(..) , EndPoint(..) @@ -29,7 +29,8 @@ import Control.Exception (Exception) import Control.Applicative ((<$>)) import Data.Typeable (Typeable) -import Data.Binary (Binary(get, put)) +import Data.Binary (Binary(put, get), putWord8, getWord8) +import Data.Hashable import Data.Word (Word64) -------------------------------------------------------------------------------- @@ -41,8 +42,8 @@ data Transport = Transport { -- | Create a new end point (heavyweight operation) newEndPoint :: IO (Either (TransportError NewEndPointErrorCode) EndPoint) - -- | Shutdown the transport completely - , closeTransport :: IO () + -- | Shutdown the transport completely + , closeTransport :: IO () } -- | Network endpoint. @@ -50,8 +51,8 @@ -- | Endpoints have a single shared receive queue. receive :: IO Event -- | EndPointAddress of the endpoint. - , address :: EndPointAddress - -- | Create a new lightweight connection. + , address :: EndPointAddress + -- | Create a new lightweight connection. -- -- 'connect' should be as asynchronous as possible; for instance, in -- Transport implementations based on some heavy-weight underlying network @@ -64,12 +65,12 @@ , resolveMulticastGroup :: MulticastAddress -> IO (Either (TransportError ResolveMulticastGroupErrorCode) MulticastGroup) -- | Close the endpoint , closeEndPoint :: IO () - } + } -- | Lightweight connection to an endpoint. data Connection = Connection { -- | Send a message on this connection. - -- + -- -- 'send' provides vectored I/O, and allows multiple data segments to be -- sent using a single call (cf. 'Network.Socket.ByteString.sendMany'). -- Note that this segment structure is entirely unrelated to the segment @@ -80,7 +81,7 @@ } -- | Event on an endpoint. -data Event = +data Event = -- | Received a message Received {-# UNPACK #-} !ConnectionId [ByteString] -- | Connection closed @@ -88,33 +89,45 @@ -- | Connection opened -- -- 'ConnectionId's need not be allocated contiguously. - | ConnectionOpened {-# UNPACK #-} !ConnectionId Reliability EndPointAddress + | ConnectionOpened {-# UNPACK #-} !ConnectionId Reliability EndPointAddress -- | Received multicast | ReceivedMulticast MulticastAddress [ByteString] -- | The endpoint got closed (manually, by a call to closeEndPoint or closeTransport) | EndPointClosed - -- | An error occurred - | ErrorEvent (TransportError EventErrorCode) + -- | An error occurred + | ErrorEvent (TransportError EventErrorCode) deriving (Show, Eq) -- | Connection data ConnectHintsIDs enable receivers to distinguish one connection from another. -type ConnectionId = Word64 +type ConnectionId = Word64 -- | Reliability guarantees of a connection. -data Reliability = - ReliableOrdered - | ReliableUnordered +data Reliability = + ReliableOrdered + | ReliableUnordered | Unreliable - deriving (Show, Eq) + deriving (Show, Eq, Typeable) +instance Binary Reliability where + put ReliableOrdered = putWord8 0 + put ReliableUnordered = putWord8 1 + put Unreliable = putWord8 2 + get = do + header <- getWord8 + case header of + 0 -> return ReliableOrdered + 1 -> return ReliableUnordered + 2 -> return Unreliable + _ -> fail "Reliability.get: invalid" + -- | Multicast group. data MulticastGroup = MulticastGroup { - -- | EndPointAddress of the multicast group. + -- | EndPointAddress of the multicast group. multicastAddress :: MulticastAddress -- | Delete the multicast group completely. , deleteMulticastGroup :: IO () -- | Maximum message size that we can send to this group. - , maxMsgSize :: Maybe Int + , maxMsgSize :: Maybe Int -- | Send a message to the group. , multicastSend :: [ByteString] -> IO () -- | Subscribe to the given multicast group (to start receiving messages from the group). @@ -127,7 +140,7 @@ -- | EndPointAddress of an endpoint. newtype EndPointAddress = EndPointAddress { endPointAddressToByteString :: ByteString } - deriving (Eq, Ord, Typeable) + deriving (Eq, Ord, Typeable, Hashable) instance Binary EndPointAddress where put = put . endPointAddressToByteString @@ -172,13 +185,13 @@ -------------------------------------------------------------------------------- -- | Errors returned by Network.Transport API functions consist of an error --- code and a human readable description of the problem +-- code and a human readable description of the problem data TransportError error = TransportError error String deriving (Show, Typeable) -- | Although the functions in the transport API never throw TransportErrors -- (but return them explicitly), application code may want to turn these into --- exceptions. +-- exceptions. instance (Typeable err, Show err) => Exception (TransportError err) -- | When comparing errors we ignore the human-readable strings @@ -190,19 +203,19 @@ -- | Not enough resources NewEndPointInsufficientResources -- | Failed for some other reason - | NewEndPointFailed + | NewEndPointFailed deriving (Show, Typeable, Eq) --- | Connection failure -data ConnectErrorCode = - -- | Could not resolve the address +-- | Connection failure +data ConnectErrorCode = + -- | Could not resolve the address ConnectNotFound -- | Insufficient resources (for instance, no more sockets available) - | ConnectInsufficientResources + | ConnectInsufficientResources -- | Timeout | ConnectTimeout -- | Failed for other reasons (including syntax error) - | ConnectFailed + | ConnectFailed deriving (Show, Typeable, Eq) -- | Failure during the creation of a new multicast group @@ -221,7 +234,7 @@ ResolveMulticastGroupNotFound -- | Failed for some other reason (including syntax error) | ResolveMulticastGroupFailed - -- | Not all transport implementations support multicast + -- | Not all transport implementations support multicast | ResolveMulticastGroupUnsupported deriving (Show, Typeable, Eq) @@ -230,12 +243,12 @@ -- | Connection was closed SendClosed -- | Send failed for some other reason - | SendFailed + | SendFailed deriving (Show, Typeable, Eq) -- | Error codes used when reporting errors to endpoints (through receive) -data EventErrorCode = - -- | Failure of the entire endpoint +data EventErrorCode = + -- | Failure of the entire endpoint EventEndPointFailed -- | Transport-wide fatal error | EventTransportFailed @@ -247,27 +260,27 @@ -- both directions, must now be considered to have failed; they fail as a -- "bundle" of connections, with only a single "bundle" of connections per -- endpoint at any point in time. - -- + -- -- That is, suppose there are multiple connections in either direction -- between endpoints A and B, and A receives a notification that it has -- lost contact with B. Then A must not be able to send any further - -- messages to B on existing connections. + -- messages to B on existing connections. -- -- Although B may not realize /immediately/ that its connection to A has -- been broken, messages sent by B on existing connections should not be -- delivered, and B must eventually get an EventConnectionLost message, - -- too. + -- too. -- -- Moreover, this event must be posted before A has successfully -- reconnected (in other words, if B notices a reconnection attempt from A, -- it must post the EventConnectionLost before acknowledging the connection -- from A) so that B will not receive events about new connections or - -- incoming messages from A without realizing that it got disconnected. - -- + -- incoming messages from A without realizing that it got disconnected. + -- -- If B attempts to establish another connection to A before it realized -- that it got disconnected from A then it's okay for this connection -- attempt to fail, and the EventConnectionLost to be posted at that point, -- or for the EventConnectionLost to be posted and for the new connection -- to be considered the first connection of the "new bundle". - | EventConnectionLost EndPointAddress + | EventConnectionLost EndPointAddress deriving (Show, Typeable, Eq) diff --git a/src/Network/Transport/Internal.hs b/src/Network/Transport/Internal.hs --- a/src/Network/Transport/Internal.hs +++ b/src/Network/Transport/Internal.hs @@ -1,5 +1,5 @@ -- | Internal functions -module Network.Transport.Internal +module Network.Transport.Internal ( -- * Encoders/decoders encodeInt32 , decodeInt32 @@ -28,13 +28,13 @@ import Foreign.ForeignPtr (withForeignPtr) import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length) -import qualified Data.ByteString.Internal as BSI +import qualified Data.ByteString.Internal as BSI ( unsafeCreate , toForeignPtr , inlinePerformIO ) import Control.Monad.IO.Class (MonadIO, liftIO) -import Control.Exception +import Control.Exception ( IOException , SomeException , AsyncException @@ -56,36 +56,36 @@ foreign import ccall unsafe "htons" htons :: CShort -> CShort foreign import ccall unsafe "ntohs" ntohs :: CShort -> CShort --- | Serialize 32-bit to network byte order +-- | Serialize 32-bit to network byte order encodeInt32 :: Enum a => a -> ByteString -encodeInt32 i32 = +encodeInt32 i32 = BSI.unsafeCreate 4 $ \p -> pokeByteOff p 0 (htonl . fromIntegral . fromEnum $ i32) --- | Deserialize 32-bit from network byte order +-- | Deserialize 32-bit from network byte order -- Throws an IO exception if this is not a valid integer. -decodeInt32 :: Num a => ByteString -> a -decodeInt32 bs - | BS.length bs /= 4 = throw $ userError "decodeInt32: Invalid length" - | otherwise = BSI.inlinePerformIO $ do - let (fp, offset, _) = BSI.toForeignPtr bs +decodeInt32 :: Num a => ByteString -> a +decodeInt32 bs + | BS.length bs /= 4 = throw $ userError "decodeInt32: Invalid length" + | otherwise = BSI.inlinePerformIO $ do + let (fp, offset, _) = BSI.toForeignPtr bs withForeignPtr fp $ \p -> do - w32 <- peekByteOff p offset + w32 <- peekByteOff p offset return (fromIntegral . ntohl $ w32) --- | Serialize 16-bit to network byte order -encodeInt16 :: Enum a => a -> ByteString -encodeInt16 i16 = +-- | Serialize 16-bit to network byte order +encodeInt16 :: Enum a => a -> ByteString +encodeInt16 i16 = BSI.unsafeCreate 2 $ \p -> pokeByteOff p 0 (htons . fromIntegral . fromEnum $ i16) --- | Deserialize 16-bit from network byte order +-- | Deserialize 16-bit from network byte order -- Throws an IO exception if this is not a valid integer decodeInt16 :: Num a => ByteString -> a -decodeInt16 bs - | BS.length bs /= 2 = throw $ userError "decodeInt16: Invalid length" +decodeInt16 bs + | BS.length bs /= 2 = throw $ userError "decodeInt16: Invalid length" | otherwise = BSI.inlinePerformIO $ do - let (fp, offset, _) = BSI.toForeignPtr bs + let (fp, offset, _) = BSI.toForeignPtr bs withForeignPtr fp $ \p -> do w16 <- peekByteOff p offset return (fromIntegral . ntohs $ w16) @@ -120,17 +120,17 @@ forkIOWithUnmask io = forkIO (io unsafeUnmask) -- | Safe version of 'toEnum' -tryToEnum :: (Enum a, Bounded a) => Int -> Maybe a +tryToEnum :: (Enum a, Bounded a) => Int -> Maybe a tryToEnum = go minBound maxBound where go :: Enum b => b -> b -> Int -> Maybe b - go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing + go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing -- | If the timeout value is not Nothing, wrap the given computation with a -- timeout and it if times out throw the specified exception. Identity -- otherwise. timeoutMaybe :: Exception e => Maybe Int -> e -> IO a -> IO a -timeoutMaybe Nothing _ f = f +timeoutMaybe Nothing _ f = f timeoutMaybe (Just n) e f = do ma <- timeout n f case ma of @@ -138,19 +138,19 @@ Just a -> return a -- | @asyncWhenCancelled g f@ runs f in a separate thread and waits for it --- to complete. If f throws an exception we catch it and rethrow it in the +-- to complete. If f throws an exception we catch it and rethrow it in the -- current thread. If the current thread is interrupted before f completes, -- we run the specified clean up handler (if f throws an exception we assume -- that no cleanup is necessary). asyncWhenCancelled :: forall a. (a -> IO ()) -> IO a -> IO a asyncWhenCancelled g f = mask_ $ do mvar <- newEmptyMVar - forkIO $ try f >>= putMVar mvar + forkIO $ try f >>= putMVar mvar -- takeMVar is interruptible (even inside a mask_) catch (takeMVar mvar) (exceptionHandler mvar) >>= either throwIO return where - exceptionHandler :: MVar (Either SomeException a) - -> AsyncException + exceptionHandler :: MVar (Either SomeException a) + -> AsyncException -> IO (Either SomeException a) exceptionHandler mvar ex = do forkIO $ takeMVar mvar >>= either (const $ return ()) g diff --git a/src/Network/Transport/Util.hs b/src/Network/Transport/Util.hs --- a/src/Network/Transport/Util.hs +++ b/src/Network/Transport/Util.hs @@ -1,9 +1,9 @@ --- | Utility functions --- +-- | Utility functions +-- -- Note: this module is bound to change even more than the rest of the API :) module Network.Transport.Util (spawn) where -import Network.Transport +import Network.Transport ( Transport , EndPoint(..) , EndPointAddress @@ -14,9 +14,9 @@ import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) -- | Fork a new thread, create a new end point on that thread, and run the specified IO operation on that thread. --- +-- -- Returns the address of the new end point. -spawn :: Transport -> (EndPoint -> IO ()) -> IO EndPointAddress +spawn :: Transport -> (EndPoint -> IO ()) -> IO EndPointAddress spawn transport proc = do addrMVar <- newEmptyMVar forkIO $ do