network-transport (empty) → 0.2.0
raw patch · 6 files changed
+513/−0 lines, 6 filesdep +basedep +binarydep +bytestringsetup-changed
Dependencies added: base, binary, bytestring, transformers
Files
- LICENSE +31/−0
- Setup.hs +2/−0
- network-transport.cabal +80/−0
- src/Network/Transport.hs +226/−0
- src/Network/Transport/Internal.hs +150/−0
- src/Network/Transport/Util.hs +24/−0
+ LICENSE view
@@ -0,0 +1,31 @@+Copyright Well-Typed LLP, 2011-2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of the owner nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ network-transport.cabal view
@@ -0,0 +1,80 @@+Name: network-transport+Version: 0.2.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, dcoutts@well-typed.com+Stability: experimental+Homepage: http://github.com/haskell-distributed/distributed-process+Bug-Reports: mailto:edsko@well-typed.com+Synopsis: Network abstraction layer +Description: "Network.Transport" is a Network Abstraction Layer which provides+ the following high-level concepts:+ .+ * Nodes in the network are represented by 'EndPoint's. These are+ heavyweight stateful objects.+ .+ * Each 'EndPoint' has an 'EndPointAddress'.+ .+ * Connections can be established from one 'EndPoint' to another+ using the 'EndPointAddress' of the remote end.+ .+ * The 'EndPointAddress' can be serialised and sent over the+ network, where as 'EndPoint's and connections cannot.+ .+ * Connections between 'EndPoint's are unidirectional and lightweight.+ .+ * Outgoing messages are sent via a 'Connection' object that+ represents the sending end of the connection.+ .+ * Incoming messages for /all/ of the incoming connections on+ an 'EndPoint' are collected via a shared receive queue.+ .+ * In addition to incoming messages, 'EndPoint's are notified of+ other 'Event's such as new connections or broken connections.+ . + This design was heavily influenced by the design of the Common+ Communication Interface+ (<http://www.olcf.ornl.gov/center-projects/common-communication-interface>).+ Important design goals are:+ . + * Connections should be lightweight: it should be no problem to+ create thousands of connections between endpoints.+ .+ * Error handling is explicit: every function declares as part of+ its type which errors it can return (no exceptions are thrown)+ .+ * Error handling is "abstract": errors that originate from+ implementation specific problems (such as "no more sockets" in+ the TCP implementation) get mapped to generic errors+ ("insufficient resources") at the Transport level. + .+ This package provides the generic interface only; you will+ 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 ++Source-Repository head+ Type: git+ Location: https://github.com/haskell-distributed/distributed-process+ SubDir: network-transport++Library+ Build-Depends: base >= 4.3 && < 5,+ binary >= 0.5 && < 0.6,+ bytestring >= 0.9 && < 0.10,+ transformers >= 0.2 && < 0.4+ Exposed-Modules: Network.Transport,+ Network.Transport.Util+ Network.Transport.Internal+ Extensions: ForeignFunctionInterface, + RankNTypes, + ScopedTypeVariables,+ DeriveDataTypeable,+ GeneralizedNewtypeDeriving+ GHC-Options: -Wall -fno-warn-unused-do-bind+ HS-Source-Dirs: src
+ src/Network/Transport.hs view
@@ -0,0 +1,226 @@+-- | Network Transport +module Network.Transport ( -- * Types+ Transport(..)+ , EndPoint(..)+ , Connection(..)+ , Event(..)+ , ConnectionId+ , Reliability(..)+ , MulticastGroup(..)+ , EndPointAddress(..)+ , MulticastAddress(..)+ -- * Hints+ , ConnectHints(..)+ , defaultConnectHints+ -- * Error codes+ , TransportError(..)+ , NewEndPointErrorCode(..)+ , ConnectErrorCode(..)+ , NewMulticastGroupErrorCode(..)+ , ResolveMulticastGroupErrorCode(..)+ , SendErrorCode(..)+ , EventErrorCode(..)+ ) where++import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as BSC (unpack)+import Control.Exception (Exception)+import Data.Typeable (Typeable)+import Data.Binary (Binary)++--------------------------------------------------------------------------------+-- Main API --+--------------------------------------------------------------------------------++-- | To create a network abstraction layer, use one of the+-- @Network.Transport.*@ packages.+data Transport = Transport {+ -- | Create a new end point (heavyweight operation)+ newEndPoint :: IO (Either (TransportError NewEndPointErrorCode) EndPoint)+ -- | Shutdown the transport completely + , closeTransport :: IO () + }++-- | Network endpoint.+data EndPoint = EndPoint {+ -- | Endpoints have a single shared receive queue.+ receive :: IO Event+ -- | EndPointAddress of the endpoint.+ , address :: EndPointAddress + -- | Create a new lightweight connection. + , connect :: EndPointAddress -> Reliability -> ConnectHints -> IO (Either (TransportError ConnectErrorCode) Connection)+ -- | Create a new multicast group.+ , newMulticastGroup :: IO (Either (TransportError NewMulticastGroupErrorCode) MulticastGroup)+ -- | Resolve an address to a multicast group.+ , 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 :: [ByteString] -> IO (Either (TransportError SendErrorCode) ())+ -- | Close the connection.+ , close :: IO ()+ }++-- | Event on an endpoint.+data Event = + -- | Received a message+ Received ConnectionId [ByteString]+ -- | Connection closed+ | ConnectionClosed ConnectionId+ -- | Connection opened+ | ConnectionOpened 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) + deriving (Show, Eq)++-- | Connection data ConnectHintsIDs enable receivers to distinguish one connection from another.+type ConnectionId = Int++-- | Reliability guarantees of a connection.+data Reliability = + ReliableOrdered + | ReliableUnordered + | Unreliable+ deriving (Show, Eq)++-- | Multicast group.+data MulticastGroup = MulticastGroup {+ -- | 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 + -- | Send a message to the group.+ , multicastSend :: [ByteString] -> IO ()+ -- | Subscribe to the given multicast group (to start receiving messages from the group).+ , multicastSubscribe :: IO ()+ -- | Unsubscribe from the given multicast group (to stop receiving messages from the group).+ , multicastUnsubscribe :: IO ()+ -- | Close the group (that is, indicate you no longer wish to send to the group).+ , multicastClose :: IO ()+ }++-- | EndPointAddress of an endpoint.+newtype EndPointAddress = EndPointAddress { endPointAddressToByteString :: ByteString }+ deriving (Eq, Ord, Typeable, Binary)++instance Show EndPointAddress where+ show = BSC.unpack . endPointAddressToByteString++-- | EndPointAddress of a multicast group.+newtype MulticastAddress = MulticastAddress { multicastAddressToByteString :: ByteString }+ deriving (Eq, Ord)++instance Show MulticastAddress where+ show = show . multicastAddressToByteString++--------------------------------------------------------------------------------+-- Hints --+-- --+-- Hints provide transport-generic "suggestions". For now, these are --+-- placeholders only. --+--------------------------------------------------------------------------------++-- | Hints used by 'connect'+data ConnectHints = ConnectHints {+ -- Timeout+ connectTimeout :: Maybe Int+ }++-- | Default hints for connecting+defaultConnectHints :: ConnectHints+defaultConnectHints = ConnectHints {+ connectTimeout = Nothing+ }++--------------------------------------------------------------------------------+-- Error codes --+-- --+-- Errors should be transport-implementation independent. The deciding factor --+-- for distinguishing one kind of error from another should be: might --+-- application code have to take a different action depending on the kind of --+-- error? --+--------------------------------------------------------------------------------++-- | Errors returned by Network.Transport API functions consist of an error+-- 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. +instance (Typeable err, Show err) => Exception (TransportError err)++-- | When comparing errors we ignore the human-readable strings+instance Eq error => Eq (TransportError error) where+ TransportError err1 _ == TransportError err2 _ = err1 == err2++-- | Errors during the creation of an endpoint+data NewEndPointErrorCode =+ -- | Not enough resources+ NewEndPointInsufficientResources+ -- | Failed for some other reason+ | NewEndPointFailed + deriving (Show, Typeable, Eq)++-- | Connection failure +data ConnectErrorCode = + -- | Could not resolve the address + ConnectNotFound+ -- | Insufficient resources (for instance, no more sockets available)+ | ConnectInsufficientResources + -- | Timeout+ | ConnectTimeout+ -- | Failed for other reasons (including syntax error)+ | ConnectFailed + deriving (Show, Typeable, Eq)++-- | Failure during the creation of a new multicast group+data NewMulticastGroupErrorCode =+ -- | Insufficient resources+ NewMulticastGroupInsufficientResources+ -- | Failed for some other reason+ | NewMulticastGroupFailed+ -- | Not all transport implementations support multicast+ | NewMulticastGroupUnsupported+ deriving (Show, Typeable, Eq)++-- | Failure during the resolution of a multicast group+data ResolveMulticastGroupErrorCode =+ -- | Multicast group not found+ ResolveMulticastGroupNotFound+ -- | Failed for some other reason (including syntax error)+ | ResolveMulticastGroupFailed+ -- | Not all transport implementations support multicast + | ResolveMulticastGroupUnsupported+ deriving (Show, Typeable, Eq)++-- | Failure during sending a message+data SendErrorCode =+ -- | Connection was closed+ SendClosed+ -- | Send failed for some other reason+ | SendFailed + deriving (Show, Typeable, Eq)++-- | Error codes used when reporting errors to endpoints (through receive)+data EventErrorCode = + -- | Failure of the entire endpoint + EventEndPointFailed+ -- | Transport-wide fatal error+ | EventTransportFailed+ -- | Some incoming connections were closed abruptly.+ -- If an endpoint address is specified, then all connections to and+ -- from that endpoint are now lost+ | EventConnectionLost (Maybe EndPointAddress) [ConnectionId] + deriving (Show, Typeable, Eq)
+ src/Network/Transport/Internal.hs view
@@ -0,0 +1,150 @@+-- | Internal functions+module Network.Transport.Internal ( -- * Encoders/decoders+ encodeInt32+ , decodeInt32+ , encodeInt16+ , decodeInt16+ , prependLength+ -- * Miscellaneous abstractions+ , mapIOException+ , tryIO+ , tryToEnum+ , timeoutMaybe+ , asyncWhenCancelled+ -- * Replicated functionality from "base"+ , void+ , forkIOWithUnmask+ -- * Debugging+ , tlog+ ) where++import Prelude hiding (catch)+import Foreign.Storable (pokeByteOff, peekByteOff)+import Foreign.C (CInt(..), CShort(..))+import Foreign.ForeignPtr (withForeignPtr)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS (length)+import qualified Data.ByteString.Internal as BSI ( unsafeCreate+ , toForeignPtr+ , inlinePerformIO)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Exception ( IOException+ , SomeException+ , AsyncException+ , Exception+ , catch+ , try+ , throw+ , throwIO+ , mask_+ )+import Control.Concurrent (ThreadId, forkIO)+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)+import GHC.IO (unsafeUnmask)+import System.Timeout (timeout)+--import Control.Concurrent (myThreadId)++foreign import ccall unsafe "htonl" htonl :: CInt -> CInt+foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt+foreign import ccall unsafe "htons" htons :: CShort -> CShort+foreign import ccall unsafe "ntohs" ntohs :: CShort -> CShort++-- | Serialize 32-bit to network byte order +encodeInt32 :: Enum a => a -> ByteString+encodeInt32 i32 = + BSI.unsafeCreate 4 $ \p ->+ pokeByteOff p 0 (htonl . fromIntegral . fromEnum $ i32)++-- | 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 + withForeignPtr fp $ \p -> do+ w32 <- peekByteOff p offset + return (fromIntegral . ntohl $ w32)++-- | 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 +-- 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" + | otherwise = BSI.inlinePerformIO $ do+ let (fp, offset, _) = BSI.toForeignPtr bs + withForeignPtr fp $ \p -> do+ w16 <- peekByteOff p offset+ return (fromIntegral . ntohs $ w16)++-- | Prepend a list of bytestrings with their total length+prependLength :: [ByteString] -> [ByteString]+prependLength bss = encodeInt32 (sum . map BS.length $ bss) : bss++-- | Translate exceptions that arise in IO computations+mapIOException :: Exception e => (IOException -> e) -> IO a -> IO a+mapIOException f p = catch p (throwIO . f)++-- | Like 'try', but lifted and specialized to IOExceptions+tryIO :: MonadIO m => IO a -> m (Either IOException a)+tryIO = liftIO . try++-- | Logging (for debugging)+tlog :: MonadIO m => String -> m ()+tlog _ = return ()+{-+tlog msg = liftIO $ do+ tid <- myThreadId+ putStrLn $ show tid ++ ": " ++ msg+-}++-- | Not all versions of "base" export 'void'+void :: Monad m => m a -> m ()+void p = p >> return ()++-- | This was introduced in "base" some time after 7.0.4+forkIOWithUnmask :: ((forall a . IO a -> IO a) -> IO ()) -> IO ThreadId+forkIOWithUnmask io = forkIO (io unsafeUnmask)++-- | Safe version of 'toEnum'+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 ++-- | 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 (Just n) e f = do+ ma <- timeout n f+ case ma of+ Nothing -> throwIO e+ 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 +-- 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 + -- takeMVar is interruptible (even inside a mask_)+ catch (takeMVar mvar) (exceptionHandler mvar) >>= either throwIO return+ where+ exceptionHandler :: MVar (Either SomeException a) + -> AsyncException + -> IO (Either SomeException a)+ exceptionHandler mvar ex = do+ forkIO $ takeMVar mvar >>= either (const $ return ()) g+ throwIO ex
+ src/Network/Transport/Util.hs view
@@ -0,0 +1,24 @@+-- | 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 ( Transport+ , EndPoint(..)+ , EndPointAddress+ , newEndPoint+ )+import Control.Concurrent (forkIO)+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 proc = do+ addr <- newEmptyMVar+ forkIO $ do+ Right endpoint <- newEndPoint transport+ putMVar addr (address endpoint)+ proc endpoint+ takeMVar addr