diff --git a/courier.cabal b/courier.cabal --- a/courier.cabal +++ b/courier.cabal @@ -2,13 +2,20 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: courier -version: 0.1.0.5 -synopsis: A message-passing library, intended for simplifying network applications +version: 0.1.0.6 +synopsis: A message-passing library for simplifying network applications description: Inspired by Erlang's simple message-passing facilities, courier provides roughly similar capabilities. Applications simply create one or more endpoints, bind each to a transport using a given name, then can freely send / receive messages to other endpoints just by referencing the name each endpoint bound to its transport. + . + Note that while the simplicity is inspired by Erlang, the actual semantics of a receive + operation are not: receive just returns the next message delivered to an endpoint by a + transport. There is no attempt to perform pattern-matching on a range of alternatives, + and thus enabling out-of-order receipt. Consequently, all messages delivered to an endpoint + will always be received in the order delivered. In this sense, endpoints are more akin to + channels in Go but without the strict typing. homepage: http://github.com/hargettp/courier license: MIT @@ -33,9 +40,12 @@ Network.Transport Network.Transport.Memory Network.Transport.TCP + Network.Transport.UDP ghc-options: -Wall - -- other-modules: + other-modules: + Network.Transport.Sockets + Network.Transport.Internal build-depends: base >=4 && <5, async, bytestring, @@ -51,18 +61,14 @@ type: exitcode-stdio-1.0 hs-source-dirs: tests main-is: Tests.hs - other-modules: TestMemory, - TestTCP build-depends: -- base modules base, -- testing support HUnit, - QuickCheck >= 2.0, test-framework, test-framework-hunit, - test-framework-quickcheck2, -- Cabal, -- 3rd party modules cereal, 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,46 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Network.Transport.Internal +-- Copyright : (c) Phil Hargett 2013 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (uses STM) +-- +-- Internal definitions used for implementing transports. +----------------------------------------------------------------------------- + +module Network.Transport.Internal ( + catchExceptions +) + +where + +-- local imports +-- external imports + +import Control.Exception + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +{-| +Silently ignore 'BlockedIndefinitelyOnSTM' and 'ThreadKilled' exceptions, as +most cases in which they occur are benign and not worth the noise. Use with +caution, as this function could mask undelrying design issues. +-} +catchExceptions :: IO () -> (SomeException -> IO ()) -> IO () +catchExceptions blk handler = do + catch (do + catch (catch blk ignoreBlocked) + ignoreThreadKilled) + handler + where + ignoreBlocked :: BlockedIndefinitelyOnSTM -> IO () + ignoreBlocked _ = return () + ignoreThreadKilled :: AsyncException -> IO () + ignoreThreadKilled e = + case e of + ThreadKilled -> return () + _ -> handler $ SomeException e diff --git a/src/Network/Transport/Memory.hs b/src/Network/Transport/Memory.hs --- a/src/Network/Transport/Memory.hs +++ b/src/Network/Transport/Memory.hs @@ -30,7 +30,6 @@ import Control.Concurrent.STM import qualified Data.Map as M -import Data.Maybe (fromJust) -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- @@ -75,8 +74,9 @@ memorySendTo :: MemoryTransport -> Name -> Message -> IO () memorySendTo transport name msg = do mailboxes <- atomically $ readTVar $ boundMailboxes transport - let mailbox = fromJust $ M.lookup name mailboxes - atomically $ writeTQueue mailbox msg + case M.lookup name mailboxes of + Just mailbox -> atomically $ writeTQueue mailbox msg + Nothing -> return () -- error $ "No mailbox for " ++ name memoryUnbind :: MemoryTransport -> Name -> IO () memoryUnbind transport name = do diff --git a/src/Network/Transport/Sockets.hs b/src/Network/Transport/Sockets.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Sockets.hs @@ -0,0 +1,325 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE DeriveGeneric #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Network.Transport.Sockets +-- Copyright : (c) Phil Hargett 2013 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (uses STM) +-- +-- Common facilities for socket-based transports, such as UDP and TCP/IP. +-- +----------------------------------------------------------------------------- + +module Network.Transport.Sockets ( + + Bindings, + + SocketTransport(..), + + Connection(..), + + IdentifyMessage(..), + + Messenger(..), + newMessenger, + addMessenger, + deliver, + closeMessenger, + + dispatcher, + sender, + socketSendTo, + receiver, + receiveSocketMessage, + receiveSocketMessages, + SocketSend, + + parseSocketAddress, + lookupAddresses, + lookupAddress + + ) where + +-- local imports + +import Network.Transport +import Network.Transport.Internal + +-- external imports +import Control.Concurrent.Async +import Control.Concurrent.STM + +import Control.Exception + +import qualified Data.ByteString as B +import qualified Data.Map as M +import Data.Serialize +import qualified Data.Set as S +import qualified Data.Text as T + +import GHC.Generics + +import Network.Simple.TCP (recv) +import Network.Socket hiding (recv,socket) + +import System.Log.Logger + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +_log :: String +_log = "transport.sockets" + +type Bindings = TVar (M.Map Name Mailbox) + +data SocketTransport = SocketTransport { + socketMessengers :: TVar (M.Map Address Messenger), + socketBindings :: Bindings, + socketConnection :: Address -> IO Connection, + socketMessenger :: Connection -> Mailbox -> IO Messenger, + socketInbound :: Mailbox, + socketDispatchers :: S.Set (Async ()), + socketResolver :: Resolver + +} + +{-| +A connection specializes the use of a transport for a particular +destination. +-} +data Connection = Connection { + connAddress :: Address, + connSocket :: TMVar Socket, + connConnect :: IO Socket, + connSend :: Socket -> B.ByteString -> IO (), + connReceive :: Socket -> Int -> IO (Maybe B.ByteString), + connClose :: IO () + } + +{-| +A messenger is a facility that actual uses the mechanisms of a transport +(and more specifically, of a connection on a transport) to deliver and receive +messages. The messenger uses 'Mailbox'es internally so that the sending/receiving +happens asynchronously, allowing applications to move on without regard for +when any send / receive action actually completes. +-} +data Messenger = Messenger { + messengerOut :: Mailbox, + messengerAddress :: Address, + messengerSender :: Async (), + messengerReceiver :: Async (), + messengerConnection :: Connection + } + +data IdentifyMessage = IdentifyMessage Address deriving (Generic) + +instance Serialize IdentifyMessage + +{-| +Parse a TCP 'Address' into its respective 'HostName' and 'PortNumber' components, on the +assumption the 'Address' has an identifer in the format @host:port@. If +the port number is missing from the supplied address, it will default to 0. If the +hostname component is missing from the identifier (e.g., just @:port@), then hostname +is assumed to be @localhost@. +-} +parseSocketAddress :: Address -> (HostName,ServiceName) +parseSocketAddress address = + let identifer = T.pack $ address + parts = T.splitOn ":" identifer + in if (length parts) > 1 then + (host $ T.unpack $ parts !! 0, port $ T.unpack $ parts !! 1) + else (host $ T.unpack $ parts !! 0, "0") + where + host h = if h == "" then + "localhost" + else h + port p = p + +lookupAddresses :: (HostName,ServiceName) -> IO [SockAddr] +lookupAddresses hostAndPort = + let (host,port) = hostAndPort + hints = defaultHints { addrFlags = [AI_ADDRCONFIG, AI_CANONNAME, AI_NUMERICSERV] } + in do + addresses <- getAddrInfo (Just hints) (Just host) (Just port) + return $ map addrAddress addresses + +lookupAddress :: (HostName,ServiceName) -> IO SockAddr +lookupAddress hostAndPort = do + addresses <- lookupAddresses hostAndPort + return $ addresses !! 0 + +type SocketSend = Socket -> B.ByteString -> IO () + +instance Show Messenger where + show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")" + +newMessenger :: Connection -> Mailbox -> IO Messenger +newMessenger conn inc = do + out <- newMailbox + sndr <- async $ sender conn out + rcvr <- async $ receiver conn inc + return Messenger { + messengerOut = out, + messengerAddress = connAddress conn, + messengerSender = sndr, + messengerReceiver = rcvr, + messengerConnection = conn + } + +addMessenger :: SocketTransport -> Address -> Messenger -> IO () +addMessenger transport address msngr = do + msngrs <- atomically $ do + modifyTVar (socketMessengers transport) $ \msngrs -> M.insert address msngr msngrs + msngrs <- readTVar (socketMessengers transport) + return msngrs + infoM _log $ "Added messenger to " ++ (show address) ++ "; messengers are " ++ (show msngrs) + +deliver :: Messenger -> Message -> IO () +deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message + +dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO () +dispatcher bindings mbox = dispatchMessages + where + dispatchMessages = catchExceptions (do + infoM _log $ "Dispatching messages" + env <- atomically $ readTQueue mbox + dispatchMessage env + dispatchMessages) + (\e -> do + warningM _log $ "Dispatch error: " ++ (show (e :: SomeException))) + dispatchMessage env = do + infoM _log $ "Dispatching message" + let envelopeOrErr = decode env + case envelopeOrErr of + Left err -> do + errorM _log $ "Error decoding message for dispatch: " ++ err + return () + Right (Envelope destination msg) -> do + atomically $ do + dests <- readTVar bindings + let maybeDest = M.lookup destination dests + case maybeDest of + Nothing -> return () + Just dest -> do + writeTQueue dest msg + return () + +sender :: Connection -> Mailbox -> IO () +sender conn mailbox = sendMessages + where + sendMessages = do + reconnect + catchExceptions (do + infoM _log $ "Waiting to send to " ++ (show $ connAddress conn) + msg <- atomically $ readTQueue mailbox + infoM _log $ "Sending message to " ++ (show $ connAddress conn) + connected <- atomically $ tryReadTMVar $ connSocket conn + case connected of + Just socket -> do + (connSend conn) socket msg + Nothing -> return () + ) + (\e -> do + warningM _log $ "Send error: " ++ (show (e :: SomeException)) + disconnect) + sendMessages + reconnect = do + -- TODO need a timeout here, in case connecting always fails + connected <- atomically $ tryReadTMVar $ connSocket conn + case connected of + Just _ -> return () + Nothing -> do + let (host,port) = parseSocketAddress $ connAddress conn + infoM _log $ "Connecting to " ++ (show host) ++ ":" ++ (show port) -- (show address) + socket <- connConnect conn + infoM _log $ "Connected to " ++ (show $ connAddress conn) + atomically $ putTMVar (connSocket conn) socket + disconnect = do + connected <- atomically $ tryTakeTMVar $ connSocket conn + case connected of + Just socket -> sClose socket + Nothing -> return () + + +socketSendTo :: SocketTransport -> Name -> Message -> IO () +socketSendTo transport name msg = do + isLocal <- local + if isLocal + then return () + else remote + where + local = do + found <- atomically $ do + bindings <- readTVar $ socketBindings transport + return $ M.lookup name bindings + case found of + Nothing -> return False + Just mbox -> do + atomically $ writeTQueue mbox msg + return True + remote = do + Just address <- resolve (socketResolver transport) name + let env = encode $ Envelope { + envelopeDestination = name, + envelopeContents = msg + } + amsngr <- atomically $ do + msngrs <- readTVar $ socketMessengers transport + return $ M.lookup address msngrs + case amsngr of + Nothing -> do + msngrs <- atomically $ readTVar $ socketMessengers transport + infoM _log $ "No messenger for " ++ (show address) ++ " in " ++ (show msngrs) + socketVar <- atomically $ newEmptyTMVar + newConn <- (socketConnection transport) address + let conn = newConn {connSocket = socketVar} + msngr <- (socketMessenger transport) conn (socketInbound transport) + addMessenger transport address msngr + deliver msngr env + return () + Just msngr -> deliver msngr env + +receiver :: Connection -> Mailbox -> IO () +receiver conn mailbox = do + socket <- atomically $ readTMVar $ connSocket conn + receiveSocketMessages socket (connAddress conn) mailbox + +receiveSocketMessages :: Socket -> Address -> Mailbox -> IO () +receiveSocketMessages sock addr mailbox = catchExceptions (do + infoM _log $ "Waiting to receive on " ++ (show addr) + maybeMsg <- receiveSocketMessage sock + infoM _log $ "Received message on " ++ (show addr) + case maybeMsg of + Nothing -> do + sClose sock + return () + Just msg -> do + atomically $ writeTQueue mailbox msg + receiveSocketMessages sock addr mailbox) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException))) + +receiveSocketMessage :: Socket -> IO (Maybe Message) +receiveSocketMessage socket = do + maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes + case maybeLen of + Nothing -> do + infoM _log $ "No length received" + return Nothing + Just len -> do + maybeMsg <- recv socket $ msgLength (decode len) + infoM _log $ "Received message" + return maybeMsg + where + msgLength (Right size) = size + msgLength (Left err) = error err + +closeMessenger :: Messenger -> IO () +closeMessenger msngr = do + cancel $ messengerSender msngr + cancel $ messengerReceiver msngr + connClose $ messengerConnection msngr 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 @@ -25,6 +25,8 @@ -- local imports import Network.Transport +import Network.Transport.Internal +import Network.Transport.Sockets -- external imports @@ -36,10 +38,7 @@ import qualified Data.Map as M import Data.Serialize import qualified Data.Set as S -import qualified Data.Text as T -import GHC.Generics - import Network.Socket (sClose,accept) import Network.Simple.TCP hiding (accept) @@ -51,18 +50,8 @@ _log :: String _log = "transport.tcp" -data TCPTransport = TCPTransport { - tcpListeners :: TVar (M.Map ServiceName Socket), - tcpMessengers :: TVar (M.Map Address Messenger), - tcpBindings :: TVar (M.Map Name Mailbox), - tcpInbound :: Mailbox, - tcpDispatchers :: S.Set (Async ()), - tcpResolver :: Resolver - } - -data IdentifyMessage = IdentifyMessage Address deriving (Generic) - -instance Serialize IdentifyMessage +tcpScheme :: Scheme +tcpScheme = "tcp" {-| Create a new 'Transport' suitable for sending messages over TCP/IP. There can @@ -71,67 +60,42 @@ correct TCP/IP addresses (or hostnames) for communication. -} newTCPTransport :: Resolver -> IO Transport -newTCPTransport resolver = do - listeners <- atomically $ newTVar M.empty +newTCPTransport resolver = do messengers <- atomically $ newTVar M.empty bindings <- atomically $ newTVar M.empty inbound <- newMailbox dispatch <- async $ dispatcher bindings inbound - let transport = TCPTransport { - tcpListeners = listeners, - tcpMessengers = messengers, - tcpBindings = bindings, - tcpInbound = inbound, - tcpDispatchers = S.fromList [dispatch], - tcpResolver = resolver + let transport = SocketTransport { + socketMessengers = messengers, + socketBindings = bindings, + socketInbound = inbound, + socketConnection = newTCPConnection, + socketMessenger = newTCPMessenger bindings resolver, + socketDispatchers = S.fromList [dispatch], + socketResolver = resolver } return Transport { scheme = tcpScheme, handles = tcpHandles transport, bind = tcpBind transport, - sendTo = tcpSendTo transport, + sendTo = socketSendTo transport, shutdown = tcpShutdown transport } --------------------------------------------------------------------------------- - -{-| -Parse a TCP 'Address' into its respective 'HostName' and 'PortNumber' components, on the -assumption the 'Address' has an identifer in the format @host:port@. If -the port number is missing from the supplied address, it will default to 0. If the -hostname component is missing from the identifier (e.g., just @:port@), then hostname -is assumed to be @localhost@. --} -parseTCPAddress :: Address -> (HostName,ServiceName) -parseTCPAddress address = - let identifer = T.pack $ address - parts = T.splitOn ":" identifer - in if (length parts) > 1 then - (host $ T.unpack $ parts !! 0, port $ T.unpack $ parts !! 1) - else (host $ T.unpack $ parts !! 0, "0") - where - host h = if h == "" then - "localhost" - else h - port p = p - -tcpScheme :: Scheme -tcpScheme = "tcp" - -tcpHandles :: TCPTransport -> Name -> IO Bool +tcpHandles :: SocketTransport -> Name -> IO Bool tcpHandles transport name = do - resolved <- resolve (tcpResolver transport) name + resolved <- resolve (socketResolver transport) name return $ isJust resolved where isJust (Just _) = True isJust _ = False -tcpBind :: TCPTransport -> Mailbox -> Name -> IO (Either String Binding) -tcpBind transport inc name = do - atomically $ modifyTVar (tcpBindings transport) $ \bindings -> +tcpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding) +tcpBind transport inc name = do + atomically $ modifyTVar (socketBindings transport) $ \bindings -> M.insert name inc bindings - Just address <- resolve (tcpResolver transport) name - let (_,port) = parseTCPAddress address + Just address <- resolve (socketResolver transport) name + let (_,port) = parseSocketAddress address listener <- async $ do infoM _log $ "Binding to address " ++ (show address) tcpListen address port @@ -142,9 +106,9 @@ where tcpListen address port = listen HostAny port $ \(socket,_) -> - catch (do + catchExceptions (do tcpAccept address socket) - (\e -> do + (\e -> do warningM _log $ "Listen error: " ++ (show (e :: SomeException))) tcpAccept address socket = do infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show socket) @@ -159,9 +123,11 @@ Just (IdentifyMessage clientAddress) -> do infoM _log $ "Identified " ++ (show clientAddress) clientSocket <- atomically $ newTMVar client - msngr <- newMessenger clientSocket clientAddress (tcpInbound transport) - found <- atomically $ do - msngrs <- readTVar $ tcpMessengers transport + newConn <- newTCPConnection clientAddress + let conn = newConn {connSocket = clientSocket} + msngr <- newMessenger conn (socketInbound transport) + found <- atomically $ do + msngrs <- readTVar $ socketMessengers transport return $ M.lookup clientAddress msngrs case found of Just _ -> do @@ -171,7 +137,7 @@ addMessenger transport clientAddress msngr tcpIdentify client clientAddress = do infoM _log $ "Awaiting identity from " ++ (show clientAddress) - maybeMsg <- receiveMessage client + maybeMsg <- receiveSocketMessage client case maybeMsg of Nothing -> return Nothing Just bytes -> do @@ -183,203 +149,55 @@ infoM _log $ "Unbinding from port " ++ (show address) cancel listener -tcpSendTo :: TCPTransport -> Name -> Message -> IO () -tcpSendTo transport name msg = do - isLocal <- local - if isLocal - then return () - else remote - where - local = do - found <- atomically $ do - bindings <- readTVar $ tcpBindings transport - return $ M.lookup name bindings - case found of - Nothing -> return False - Just mbox -> do - atomically $ writeTQueue mbox msg - return True - remote = do - Just address <- resolve (tcpResolver transport) name - let env = encode $ Envelope { - envelopeDestination = name, - envelopeContents = msg - } - amsngr <- atomically $ do - msngrs <- readTVar $ tcpMessengers transport - return $ M.lookup address msngrs - case amsngr of - Nothing -> do - msngrs <- atomically $ readTVar $ tcpMessengers transport - infoM _log $ "No messenger for " ++ (show address) ++ " in " ++ (show msngrs) - socketVar <- atomically $ newEmptyTMVar - msngr <- newMessenger socketVar address (tcpInbound transport) - addMessenger transport address msngr - identifyAll msngr - deliver msngr env - return () - Just msngr -> deliver msngr env - deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message - identifyAll msngr = do - bindings <- atomically $ readTVar $ tcpBindings transport - boundAddresses <- mapM (resolve $ tcpResolver transport) (M.keys bindings) - let uniqueAddresses = S.toList $ S.fromList boundAddresses - mapM_ (identify msngr) uniqueAddresses - identify msngr maybeUniqueAddress= do - case maybeUniqueAddress of - Nothing -> return() - Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress - -tcpShutdown :: TCPTransport -> IO () -tcpShutdown transport = do - infoM _log $ "Closing messengers" - msngrs <- atomically $ readTVar $ tcpMessengers transport - mapM_ closeMessenger $ M.elems msngrs - infoM _log $ "Closing listeners" - listeners <- atomically $ readTVar $ tcpListeners transport - mapM_ sClose $ M.elems listeners - infoM _log $ "Closing dispatcher" - mapM_ cancel $ S.toList $ tcpDispatchers transport - mapM_ wait $ S.toList $ tcpDispatchers transport - -data Messenger = Messenger { - messengerOut :: Mailbox, - messengerAddress :: Address, - messengerSender :: Async (), - messengerReceiver :: Async (), - messengerSocket :: TMVar Socket - } - -instance Show Messenger where - show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")" - -newMessenger :: TMVar Socket -> Address -> Mailbox -> IO Messenger -newMessenger socket address inc = do - out <- newMailbox - sndr <- async $ sender socket address out - rcvr <- async $ receiver socket address inc - return Messenger { - messengerOut = out, - messengerAddress = address, - messengerSender = sndr, - messengerReceiver = rcvr, - messengerSocket = socket +newTCPConnection :: Address -> IO Connection +newTCPConnection address = do + sock <- atomically $ newEmptyTMVar + let (host,port) = parseSocketAddress address + return Connection { + connAddress = address, + connSocket = sock, + connConnect = do + (s,_) <- connectSock host port + return s, + -- connSend = send, + connSend = tcpSend address, + connReceive = recv, + connClose = do + open <- atomically $ tryTakeTMVar sock + case open of + Just socket -> sClose socket + Nothing -> return () } - -addMessenger :: TCPTransport -> Address -> Messenger -> IO () -addMessenger transport address msngr = do - msngrs <- atomically $ do - modifyTVar (tcpMessengers transport) $ \msngrs -> M.insert address msngr msngrs - msngrs <- readTVar (tcpMessengers transport) - return msngrs - infoM _log $ "Added messenger to " ++ (show address) ++ "; messengers are " ++ (show msngrs) -closeMessenger :: Messenger -> IO () -closeMessenger msngr = do - cancel $ messengerSender msngr - cancel $ messengerReceiver msngr - open <- atomically $ tryTakeTMVar $ messengerSocket msngr - case open of - Just socket -> sClose socket - Nothing -> return () - +newTCPMessenger :: Bindings -> Resolver -> Connection -> Mailbox -> IO Messenger +newTCPMessenger bindings resolver conn mailbox = do + msngr <- newMessenger conn mailbox + identifyAll msngr + return msngr + where + identifyAll msngr = do + bs <- atomically $ readTVar bindings + boundAddresses <- mapM (resolve resolver) (M.keys bs) + let uniqueAddresses = S.toList $ S.fromList boundAddresses + mapM_ (identify msngr) uniqueAddresses + identify msngr maybeUniqueAddress= do + case maybeUniqueAddress of + Nothing -> return() + Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress -sender :: TMVar Socket -> Address -> Mailbox -> IO () -sender socketVar address mailbox = sendMessages - where - sendMessages = do - reconnect - catch (do - infoM _log $ "Waiting to send to " ++ (show address) - msg <- atomically $ readTQueue mailbox - infoM _log $ "Sending message to " ++ (show address) - connected <- atomically $ tryReadTMVar socketVar - case connected of - Just socket -> do - send socket $ encode (B.length msg) - infoM _log $ "Length sent" - send socket msg - infoM _log $ "Message sent to" ++ (show address) - Nothing -> return () - ) (\e -> do - warningM _log $ "Send error: " ++ (show (e :: SomeException)) - disconnect) - sendMessages - reconnect = do - -- TODO need a timeout here, in case connecting always fails - connected <- atomically $ tryReadTMVar socketVar - case connected of - Just _ -> return () - Nothing -> do - let (host,port) = parseTCPAddress address - infoM _log $ "Connecting to " ++ (show host) ++ ":" ++ (show port) -- (show address) - (socket,sockAddr) <- connectSock host port - infoM _log $ "Connected to " ++ (show address) ++ ": " ++ (show sockAddr) - atomically $ putTMVar socketVar socket - disconnect = do - connected <- atomically $ tryTakeTMVar socketVar - case connected of - Just socket -> sClose socket - Nothing -> return () +tcpSend :: Address -> Socket -> B.ByteString -> IO () +tcpSend addr sock bs = do + send sock $ encode (B.length bs) + infoM _log $ "Length sent" + send sock bs + infoM _log $ "Message sent to" ++ (show addr) -dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO () -dispatcher bindings mbox = dispatchMessages - where - dispatchMessages = catch (do - infoM _log $ "Dispatching messages" - env <- atomically $ readTQueue mbox - dispatchMessage env - dispatchMessages) - (\e -> do - warningM _log $ "Dispatch error: " ++ (show (e :: SomeException))) - dispatchMessage env = do - infoM _log $ "Dispatching message" - let envelopeOrErr = decode env - case envelopeOrErr of - Left err -> do - errorM _log $ "Error decoding message for dispatch: " ++ err - return () - Right (Envelope destination msg) -> do - atomically $ do - dests <- readTVar bindings - let maybeDest = M.lookup destination dests - case maybeDest of - Nothing -> return () - Just dest -> do - writeTQueue dest msg - return () - -receiver :: TMVar Socket -> Address -> Mailbox -> IO () -receiver socketVar address mailbox = receiveMessages - where - receiveMessages = catch (do - infoM _log $ "Waiting to receive from " ++ (show address) - socket <- atomically $ readTMVar socketVar - maybeMsg <- receiveMessage socket - infoM _log $ "Received message from " ++ (show address) - case maybeMsg of - Nothing -> do - sClose socket - return () - Just msg -> do - atomically $ writeTQueue mailbox msg - receiveMessages) (\e -> do - warningM _log $ "Receive error: " ++ (show (e :: SomeException))) - -receiveMessage :: Socket -> IO (Maybe Message) -receiveMessage socket = catch (do - maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes - case maybeLen of - Nothing -> do - errorM _log $ "No length received" - return Nothing - Just len -> do - maybeMsg <- recv socket $ msgLength (decode len) - infoM _log $ "Received message" - return maybeMsg) (\e -> do - warningM _log $ "Receive error: " ++ (show (e :: SomeException)) - throw e) - where - msgLength (Right size) = size - msgLength (Left err) = error err +tcpShutdown :: SocketTransport -> IO () +tcpShutdown transport = do + infoM _log $ "Closing messengers" + msngrs <- atomically $ readTVar $ socketMessengers transport + mapM_ closeMessenger $ M.elems msngrs + infoM _log $ "Closing dispatcher" + mapM_ cancel $ S.toList $ socketDispatchers transport + mapM_ wait $ S.toList $ socketDispatchers transport diff --git a/src/Network/Transport/UDP.hs b/src/Network/Transport/UDP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/UDP.hs @@ -0,0 +1,181 @@ +{-# LANGUAGE DeriveGeneric #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Network.Transport.UDP +-- Copyright : (c) Phil Hargett 2013 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (uses STM) +-- +-- UDP transports deliver messages to other 'Network.Endpoints.Endpoint's using UDP/IP. +-- +-- Each UDP transport manages socket bindings on behalf of +-- 'Endpoint's, dynamically opening / closing new sockets as needed to deliver +-- messages to other 'Endpoint's using UDP transports. +-- +-- There is no reuse of sockets on the sending side, so while messages will be +-- received on a known bound port, the remote sending port will vary arbitrarily. +-- +-- This transport only reads at most 512 bytes from incoming packets: constraining +-- the packet size avoids fragmentation. Applications using this transport should take +-- responsibility for fragmentation, reassembly, retransmission of lost packets, +-- and congestion control. +----------------------------------------------------------------------------- + +module Network.Transport.UDP ( + newUDPTransport + ) where + +-- local imports + +import Network.Transport +import Network.Transport.Internal +import Network.Transport.Sockets + +-- external imports + +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Exception + +import qualified Data.ByteString as B +import qualified Data.Map as M +import qualified Data.Set as S + +import qualified Network.Socket as N +import Network.Socket.ByteString(sendAllTo,recvFrom) + +import System.Log.Logger +-- import qualified System.Random as R + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +_log :: String +_log = "transport.udp" + +udpScheme :: Scheme +udpScheme = "udp" + +newUDPTransport :: Resolver -> IO Transport +newUDPTransport resolver = do + messengers <- atomically $ newTVar M.empty + bindings <- atomically $ newTVar M.empty + inbound <- newMailbox + dispatch <- async $ dispatcher bindings inbound + let transport = SocketTransport { + socketMessengers = messengers, + socketBindings = bindings, + socketInbound = inbound, + socketConnection = newUDPConnection, + socketMessenger = newUDPMessenger, + socketDispatchers = S.fromList [dispatch], + socketResolver = resolver + } + return Transport { + scheme = udpScheme, + handles = udpHandles transport, + bind = udpBind transport, + sendTo = socketSendTo transport, + shutdown = udpShutdown transport + } + +udpHandles :: SocketTransport -> Name -> IO Bool +udpHandles transport name = do + resolved <- resolve (socketResolver transport) name + return $ isJust resolved + where + isJust (Just _) = True + isJust _ = False + +udpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding) +udpBind transport inc name = do + atomically $ modifyTVar (socketBindings transport) $ \bindings -> + M.insert name inc bindings + Just address <- resolve (socketResolver transport) name + let (_,port) = parseSocketAddress address + addrinfos <- N.getAddrInfo + (Just (N.defaultHints {N.addrFlags = [N.AI_PASSIVE]})) + Nothing (Just port) + let serveraddr = head addrinfos + sock <- N.socket (N.addrFamily serveraddr) N.Datagram N.defaultProtocol + -- have to set this option in case we frequently rebind sockets + infoM _log $ "Binding to " ++ (show port) ++ " over UDP" + N.setSocketOption sock N.ReuseAddr 1 + N.bindSocket sock $ N.addrAddress serveraddr + infoM _log $ "Bound to " ++ (show port) ++ " over UDP" + rcvr <- async $ udpReceiveSocketMessages sock address (socketInbound transport) + return $ Right Binding { + bindingName = name, + unbind = do + cancel rcvr + infoM _log $ "Closing UDP socket bound on " ++ (show port) + N.sClose sock + } + +newUDPConnection :: Address -> IO Connection +newUDPConnection address = do + sock <- atomically newEmptyTMVar + return Connection { + connAddress = address, + connSocket = sock, + connConnect = N.socket N.AF_INET N.Datagram N.defaultProtocol, + connSend = (\s bs -> do + addr <- lookupAddress $ parseSocketAddress address + infoM _log $ "Sending via UDP to " ++ (show addr) + sendAllTo s bs addr + infoM _log $ "Sent via UDP to " ++ (show addr)), + connReceive = udpRecvFrom, + connClose = do + maybeSocket <- atomically $ tryTakeTMVar sock + case maybeSocket of + Just s -> N.sClose s + Nothing -> return () + return () + } + +newUDPMessenger :: Connection -> Mailbox -> IO Messenger +newUDPMessenger conn mailbox = do + msngr <- newMessenger conn mailbox + return msngr + +udpReceiveSocketMessages :: N.Socket -> Address -> Mailbox -> IO () +udpReceiveSocketMessages sock addr mailbox = catchExceptions (do + infoM _log $ "Waiting to receive via UDP on " ++ (show addr) + maybeMsg <- udpReceiveSocketMessage + infoM _log $ "Received message via UDP on " ++ (show addr) + case maybeMsg of + Nothing -> do + N.sClose sock + return () + Just msg -> do + atomically $ writeTQueue mailbox msg + udpReceiveSocketMessages sock addr mailbox) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException))) + where + udpReceiveSocketMessage = do + maybeMsg <- udpRecvFrom sock 512 + infoM _log $ "Received message" + return maybeMsg + +udpRecvFrom :: N.Socket -> Int -> IO (Maybe B.ByteString) +udpRecvFrom sock count = do + (bs,addr) <- recvFrom sock count + infoM _log $ "Received UDP message from " ++ (show addr) ++ ": " ++ (show bs) + if B.null bs + then return Nothing + else return $ Just bs + +udpShutdown :: SocketTransport -> IO () +udpShutdown transport = do + infoM _log $ "Unbinding transport" + infoM _log $ "Closing messengers" + msngrs <- atomically $ readTVar $ socketMessengers transport + mapM_ closeMessenger $ M.elems msngrs + infoM _log $ "Closing dispatcher" + mapM_ cancel $ S.toList $ socketDispatchers transport + mapM_ wait $ S.toList $ socketDispatchers transport + diff --git a/tests/TestMemory.hs b/tests/TestMemory.hs deleted file mode 100644 --- a/tests/TestMemory.hs +++ /dev/null @@ -1,76 +0,0 @@ -module TestMemory (tests) where - --- local imports - -import Network.Endpoints -import Network.Transport.Memory - --- external imports - -import Data.Serialize - -import Test.Framework -import Test.HUnit -import Test.Framework.Providers.HUnit --- import Test.Framework.Providers.QuickCheck2 - ------------------------------------------------------------------------------ ------------------------------------------------------------------------------ - -tests :: [Test.Framework.Test] -tests = - [ - testCase "mem-endpoints+transport" testEndpointTransport, - testCase "mem-bind" testEndpointBind, - testCase "mem-unbind" testEndpointBindUnbind, - testCase "mem-sendReceive" testEndpointSendReceive, - testCase "mem-transport" testMemoryTransport - ] - -testEndpointTransport :: Assertion -testEndpointTransport = do - transport <- newMemoryTransport - _ <- newEndpoint [transport] - return () - -testEndpointBind :: Assertion -testEndpointBind = do - let name1 = "endpoint1" - transport <- newMemoryTransport - endpoint <- newEndpoint [transport] - Right () <- bindEndpoint endpoint name1 - return () - -testEndpointBindUnbind :: Assertion -testEndpointBindUnbind = do - let name1 = "endpoint1" - transport <- newMemoryTransport - endpoint <- newEndpoint [transport] - Right () <- bindEndpoint endpoint name1 - unbound <- unbindEndpoint endpoint name1 - case unbound of - Left err -> assertFailure $ "Unbind failed: " ++ err - Right () -> assertBool "Unbind succeeded" True - return () - -testEndpointSendReceive :: Assertion -testEndpointSendReceive = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - _ <- sendMessage endpoint1 name2 $ encode "hello!" - msg <- receiveMessage endpoint2 - assertEqual "Received message not same as sent" (Right "hello!") (decode msg) - return () - --- Memory tests - -testMemoryTransport :: Assertion -testMemoryTransport = do - _ <- newMemoryTransport - return () - diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs deleted file mode 100644 --- a/tests/TestTCP.hs +++ /dev/null @@ -1,223 +0,0 @@ -module TestTCP (tests) where - --- local imports - -import Network.Endpoints -import Network.Transport.TCP - --- external imports - -import Control.Concurrent -import Control.Exception - -import Data.Serialize - -import System.Log.Logger - -import Test.Framework -import Test.HUnit -import Test.Framework.Providers.HUnit --- import Test.Framework.Providers.QuickCheck2 - ------------------------------------------------------------------------------ ------------------------------------------------------------------------------ - -_log :: String -_log = "test.transport.tcp" - -testDelay :: Int -testDelay = 1 * 1000000 - -address1 :: Address -address1 = "localhost:2001" - -address2 :: Address -address2 = "localhost:2002" - -tests :: [Test.Framework.Test] -tests = - [ - testCase "tcp-endpoints+transport" testEndpointTransport, - testCase "tcp-bind-unbind" testEndpointBindUnbind, - testCase "tcp-send-receive" testEndpointSendReceive, - testCase "tcp-double-send-receive" testEndpointDoubleSendReceive, - testCase "tcp-send-receive-reply" testEndpointSendReceiveReply, - testCase "tcp-multiple-send-receive-reply" testEndpointMultipleSendReceiveReply, - testCase "tcp-local-send-receive-reply" testEndpointLocalSendReceiveReply - ] - -testEndpointTransport :: Assertion -testEndpointTransport = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTCPTransport resolver) - shutdown - (\transport -> do - _ <- newEndpoint [transport] - return ()) - -testEndpointBindUnbind :: Assertion -testEndpointBindUnbind = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTCPTransport resolver) - shutdown - (\transport -> do - endpoint <- newEndpoint [transport] - Right () <- bindEndpoint endpoint name1 - unbound <- unbindEndpoint endpoint name1 - case unbound of - Left err -> assertFailure $ "Unbind failed: " ++ err - Right () -> assertBool "Unbind succeeded" True - return ()) - -testEndpointSendReceive :: Assertion -testEndpointSendReceive = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTCPTransport resolver) - shutdown - (\transport1 -> - bracket (newTCPTransport resolver) - shutdown - (\transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - threadDelay testDelay - - infoM _log "Sending message from 1 to 2" - sendMessage_ endpoint1 name2 $ encode "hello!" - Just msg <- receiveMessageTimeout endpoint2 testDelay - assertEqual "Received message not same as sent" (Right "hello!") (decode msg) - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - return ())) - -testEndpointDoubleSendReceive :: Assertion -testEndpointDoubleSendReceive = do - let name1 = "endpoint1" - name2 = "endpoint2" - name3 = "endpoint3" - let resolver = resolverFromList [(name1,address1), - (name2,address2), - (name3,address1)] - bracket (newTCPTransport resolver) - shutdown - (\transport1 -> do - bracket (newTCPTransport resolver) - shutdown - (\transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - endpoint3 <- newEndpoint [transport1] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - Right () <- bindEndpoint endpoint3 name3 - threadDelay testDelay - - verifiedSend endpoint1 endpoint2 name1 name2 "hello" - verifiedSend endpoint3 endpoint2 name3 name2 "ciao!" - - Right () <- unbindEndpoint endpoint1 name1 - - verifiedSend endpoint3 endpoint2 name3 name2 "hi!" - - Right () <- unbindEndpoint endpoint2 name2 - Right () <- unbindEndpoint endpoint3 name3 - return ())) - -testEndpointSendReceiveReply :: Assertion -testEndpointSendReceiveReply = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTCPTransport resolver) - shutdown - (\transport1 -> - bracket (newTCPTransport resolver) - shutdown - (\transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - threadDelay testDelay - - verifiedSend endpoint1 endpoint2 name1 name2 "hello" - verifiedSend endpoint2 endpoint1 name2 name1 "hi!" - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - - return ())) - -testEndpointLocalSendReceiveReply :: Assertion -testEndpointLocalSendReceiveReply = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address1)] - bracket (newTCPTransport resolver) - shutdown - (\transport1 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport1] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - threadDelay testDelay - - verifiedSend endpoint1 endpoint2 name1 name2 "hello" - verifiedSend endpoint2 endpoint1 name2 name1 "hi!" - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - - return ()) - -testEndpointMultipleSendReceiveReply :: Assertion -testEndpointMultipleSendReceiveReply = do - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTCPTransport resolver) - shutdown - (\transport1 -> - bracket (newTCPTransport resolver) - shutdown - (\transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - threadDelay testDelay - - roundtrip endpoint1 endpoint2 name1 name2 - roundtrip endpoint2 endpoint1 name2 name1 - roundtrip endpoint1 endpoint2 name1 name2 - roundtrip endpoint2 endpoint1 name2 name1 - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - - return ())) - where - roundtrip endpoint1 endpoint2 name1 name2 = do - verifiedSend endpoint1 endpoint2 name1 name2 "hello" - verifiedSend endpoint2 endpoint1 name2 name1 "hi!" - -verifiedSend :: Endpoint -> Endpoint -> Name -> Name -> String -> Assertion -verifiedSend endpoint1 endpoint2 name1 name2 msg = do - infoM _log $ "Sending message from " ++ name1 ++ " to " ++ name2 - sendMessage_ endpoint1 name2 $ encode msg - Just msg1 <- receiveMessageTimeout endpoint2 testDelay - assertEqual "Received message not same as sent" (Right msg) (decode msg1) diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -13,11 +13,11 @@ import Test.Framework import Test.HUnit import Test.Framework.Providers.HUnit -import Test.Framework.Providers.QuickCheck2 -- Test modules import qualified TestMemory as M import qualified TestTCP as T +import qualified TestUDP as U ----------------------------------------------------------------------------- ----------------------------------------------------------------------------- @@ -26,29 +26,27 @@ main = do initLogging defaultMain tests - -initLogging :: IO () + +initLogging :: IO () initLogging = do let logFile = "tests.log" exists <- doesFileExist logFile - if exists - then removeFile logFile + if exists + then removeFile logFile else return () s <- fileHandler logFile INFO updateGlobalLogger rootLoggerName (setLevel WARNING) updateGlobalLogger rootLoggerName (addHandler s) tests :: [Test.Framework.Test] -tests = +tests = [ testCase "hunit" (assertBool "HUnit assertion of truth is false" True), - testProperty "quickcheck" True, testCase "endpoints" testEndpoint - ] + ] ++ M.tests ++ T.tests - -- ++ D.tests - -- ++ S.tests + ++ U.tests -- Endpoint tests