diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/network-transport.cabal b/network-transport.cabal new file mode 100644 --- /dev/null +++ b/network-transport.cabal @@ -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 + (). + 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 diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport.hs @@ -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) diff --git a/src/Network/Transport/Internal.hs b/src/Network/Transport/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Internal.hs @@ -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 diff --git a/src/Network/Transport/Util.hs b/src/Network/Transport/Util.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Util.hs @@ -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