diff --git a/changes.md b/changes.md --- a/changes.md +++ b/changes.md @@ -1,3 +1,26 @@ +0.1.0.8 + + * Enabled selective out of order mesage reception, by using a Mailbox, an extension of STM's + TQueues that extracts the next message from the queue that matches a supplied selection + function regardless of message order in the queue. + + With selective out of order message reception, endpoints can approximate the + Erlang-style of message delivery and composing applications that must handle multiple + message types becomes simpler. If an application needs to handle multiple message types, + it can choose to run separate separate message pumps concurrently, each handling different + types of messages. Each message pump can operate independently of one another. + + * Fixed issue #2 and other interim bugs all resulting from multiple name bindings that resolve + the same underlying address. Now, bound sockets are reused (with reference counting) so + that if there are multiple bindings to the same address only 1 socket is created and user. + + * Unit tests passing on Mac OS X again, mostly due to correct management of sockets and + appropriate reuse. + +0.1.0.7 + + * First inclusion of changelog in package + 0.1.0.6 * Added simple implementation of transport for UDP diff --git a/courier.cabal b/courier.cabal --- a/courier.cabal +++ b/courier.cabal @@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: courier -version: 0.1.0.7 +version: 0.1.0.8 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, @@ -10,12 +10,11 @@ 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. + Applications may process messages in the order received at an endpoint, or use + selective message reception to process the first message arriving at an endpoint + that also matches a provided selection function. Through selective message reception, + applications may approximate the style of an Erlang application, and enjoy better composability + of message reception with multiple independent dispatch routines or message pumps. extra-source-files: changes.md homepage: http://github.com/hargettp/courier @@ -23,7 +22,7 @@ license-file: LICENSE author: Phil Hargett maintainer: phil@haphazardhouse.net -copyright: Copyright (c) 2013 Phil Hargett +copyright: Copyright (c) 2013-14 Phil Hargett category: Network,Distributed Computing,Message-Oriented build-type: Simple @@ -42,6 +41,7 @@ Network.Transport.Memory Network.Transport.TCP Network.Transport.UDP + Control.Concurrent.Mailbox ghc-options: -Wall other-modules: @@ -72,10 +72,11 @@ test-framework-hunit, -- Cabal, -- 3rd party modules + async, cereal, directory, hslogger, - network-simple, + stm, -- this project's modules courier diff --git a/src/Control/Concurrent/Mailbox.hs b/src/Control/Concurrent/Mailbox.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Mailbox.hs @@ -0,0 +1,273 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.Mailbox +-- Copyright : (c) Phil Hargett 2014 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (uses STM) +-- +-- A 'Mailbox' is a drop-in replacement for 'TQueue' in "Control.Concurrent.STM", except that +-- it also supports selective out of order message reception: that is, it allows the caller +-- to dequeue the first message among the messages available in the queue that matches +-- a supplied test function, or block if no such match is possible with the messages currently +-- in the queue. +-- +-- As 'Mailbox' implements the same basic @read@/@write/@peek@ functions as a 'TQueue', +-- it offers a superset of 'TQueue' functionality, extending it with the @find@/@select@/@handle@ +-- groups of functions. Thus, applications can safely use 'Mailbox'es in place of 'TQueue's, +-- but choose when to take the slight extra overhead of 'Mailbox' functionality. +-- +-- Because message selection in worst case requires fully traversing all messages in the queue, +-- application designers are encouraged to understand this aspect when choosing to use 'Mailbox'es +-- in their designs, or when using the additional features of 'Mailbox'es beyond that of 'TQueue's. +-- +-- Despite this extra cost, 'Mailbox'es offer advantages to designers: +-- +-- * Implementation of Erlang-style message reception: as messages can be received +-- out of order, a mailbox is analogous to a process input queue in Erlang. +-- +-- * Better composability: if applications must only dequeue messages in the order in which +-- they are queued (which is sufficient for many applications), then the main message +-- pump requires modification each time a new class of message must be handled. With +-- selective message reception, multiple concurrent message pumps are possible (with +-- a small performance impact), each processing the messages they expect and with no +-- need to be aware of other message pumps performing their own work on the same mailbox. +-- +-- Basic framework for 'Mailbox' brazenly copied from "Control.Concurrent.STM.TQueue". +----------------------------------------------------------------------------- + +module Control.Concurrent.Mailbox ( + -- * Mailbox + Mailbox, + newMailbox, + newMailboxIO, + writeMailbox, + readMailbox, + tryReadMailbox, + peekMailbox, + tryPeekMailbox, + selectMailbox, + trySelectMailbox, + handleMailbox, + findMailbox, + tryFindMailbox, + unGetMailbox, + isEmptyMailbox, +) where + +-- local imports + +-- external imports + +import Control.Concurrent.STM + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +data Mailbox m = Mailbox + {-# UNPACK #-} !(TVar [m]) + {-# UNPACK #-} !(TVar [m]) + +instance Eq (Mailbox m) where + Mailbox a _ == Mailbox b _ = a == b + +{-| +Build and returns a new instance of 'Mailbox' +-} +newMailbox :: STM (Mailbox m) +newMailbox = do + _read <- newTVar [] + _write <- newTVar [] + return (Mailbox _read _write) + +{-| +@IO@ version of 'newMailbox'. This is useful for creating top-level +'Mailbox's using 'System.IO.Unsafe.unsafePerformIO', because using +'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't +possible. +-} +newMailboxIO :: IO (Mailbox m) +newMailboxIO = do + _read <- newTVarIO [] + _write <- newTVarIO [] + return (Mailbox _read _write) + +{-| +Write a value to a 'Mailbox'. +-} +writeMailbox :: Mailbox m -> m -> STM () +writeMailbox (Mailbox _read _write) msg = do + listend <- readTVar _write + writeTVar _write (msg:listend) + +{-| +Read the next value from the 'Mailbox'. +-} +readMailbox :: Mailbox m -> STM m +readMailbox (Mailbox _read _write) = do + xs <- readTVar _read + case xs of + (x:xs') -> do writeTVar _read xs' + return x + [] -> do ys <- readTVar _write + case ys of + [] -> retry + _ -> case reverse ys of + [] -> error "readMailbox" + (z:zs) -> do writeTVar _write [] + writeTVar _read zs + return z + +{-| +A version of 'readMailbox' which does not retry. Instead it +returns @Nothing@ if no value is available. +-} +tryReadMailbox :: Mailbox m -> STM (Maybe m) +tryReadMailbox mailbox = fmap Just (readMailbox mailbox) `orElse` return Nothing + +{-| +Get the next value from the @Mailbox@ without removing it, +retrying if the channel is empty. +-} +peekMailbox :: Mailbox m -> STM m +peekMailbox mailbox = do + msg <- readMailbox mailbox + unGetMailbox mailbox msg + return msg + +{-| +A version of 'peekMailbox' which does not retry. Instead it +returns @Nothing@ if no value is available. +-} +tryPeekMailbox :: Mailbox m -> STM (Maybe m) +tryPeekMailbox mailbox = do + maybeMsg <- tryReadMailbox mailbox + case maybeMsg of + Nothing -> return Nothing + Just msg -> do + unGetMailbox mailbox msg + return $ Just msg + +{-| +Find the next message in the mailbox that matches the supplied test +function or block until there is a message that does. When a message +matches (e.g., test functions returns @Just v@), return it. +-} +selectMailbox :: Mailbox m -> (m -> Maybe v) -> STM v +selectMailbox (Mailbox _read _write) testFn = do + readMessages <- readTVar _read + let (maybeReadMsg,newRead) = extract testFn readMessages + case maybeReadMsg of + Just msg -> do + writeTVar _read newRead + return msg + Nothing -> do + writeMessages <- readTVar _write + let (maybeWriteMsg,newWrite) = extract testFn writeMessages + case maybeWriteMsg of + Just msg -> do + writeTVar _write newWrite + return msg + Nothing -> retry + +{-| +A version of 'selectMailbox' which does not retry. Instead it +returns @Nothing@ if no value is available. +-} +trySelectMailbox :: Mailbox m -> (m -> Maybe v)-> STM (Maybe v) +trySelectMailbox mailbox testFn = fmap Just (selectMailbox mailbox testFn) `orElse` return Nothing + +{-| +Wait until there is a message in the mailbox matching the supplied test +function (using `selectMailbox`), then when a message is found, handle +it in the `IO` monad with the supplied function. +-} +handleMailbox :: Mailbox m -> (m -> Maybe v) -> (v -> IO r) -> IO r +handleMailbox mailbox testFn handleFn = do + selectedMsg <- atomically $ selectMailbox mailbox testFn + handleFn selectedMsg + +{-| +Find the next value from the @Mailbox@ matching @testFn@ without removing it, +retrying if the channel is empty. +-} +findMailbox :: Mailbox m -> (m -> Maybe v) -> STM v +findMailbox (Mailbox _read write) testFn = do + readMessages <- readTVar _read + case find testFn readMessages of + Just msg -> do + return msg + Nothing -> do + writeMessages <- readTVar write + case find testFn writeMessages of + Just msg -> do + return msg + Nothing -> retry + +{-| +A version of 'findMailbox' which does not retry. Instead it +returns @Nothing@ if no value is available. +-} +tryFindMailbox :: Mailbox m -> (m -> Maybe v) -> STM (Maybe v) +tryFindMailbox (Mailbox _read write) testFn = do + readMessages <- readTVar _read + case find testFn readMessages of + Just msg -> do + return $ Just msg + Nothing -> do + writeMessages <- readTVar write + case find testFn writeMessages of + Just msg -> do + return $ Just msg + Nothing -> return Nothing + +{-| +Put a data item back onto a channel, where it will be the next item read. +-} +unGetMailbox :: Mailbox m -> m -> STM () +unGetMailbox (Mailbox _read _write) msg = do + xs <- readTVar _read + writeTVar _read (msg:xs) + +{-| +Returns 'True' if the supplied 'Mailbox' is empty. +-} +isEmptyMailbox :: Mailbox m -> STM Bool +isEmptyMailbox (Mailbox _read write) = do + xs <- readTVar _read + case xs of + (_:_) -> return False + [] -> do ys <- readTVar write + case ys of + [] -> return True + _ -> return False + +-- +-- Internal helpers +-- + +{-| +Extract the first element from a list matching the +provided test and return a new list without the matching +element. +-} +extract :: (m -> Maybe v) -> [m] -> (Maybe v,[m]) +extract _ [] = (Nothing,[]) +extract test (x:xs) = + case test x of + Nothing -> let (result,rest) = extract test xs + in (result,x:rest) + Just v -> (Just v,xs) + +{-| +Find the first element from a list matching the +provided test, or Nothing if there is no match. +-} +find :: (m -> Maybe v) -> [m] -> Maybe v +find _ [] = Nothing +find testFn (x:xs) = + case testFn x of + Nothing -> find testFn xs + Just v -> Just v diff --git a/src/Network/Endpoints.hs b/src/Network/Endpoints.hs --- a/src/Network/Endpoints.hs +++ b/src/Network/Endpoints.hs @@ -34,9 +34,16 @@ sendMessage, sendMessage_, broadcastMessage, - broadcastMessage_, + broadcastMessage_, receiveMessage, receiveMessageTimeout, + postMessage, + + -- * Selective message reception + selectMessage, + selectMessageTimeout, + dispatchMessage, + dispatchMessageTimeout, -- * Transports {-| @@ -97,7 +104,7 @@ data Endpoint = Endpoint { endpointTransports :: TVar [Transport], endpointBindings :: TVar (M.Map Name Binding), - endpointMailbox :: Mailbox + endpointMailbox :: Mailbox Message } {-| @@ -107,7 +114,7 @@ newEndpoint trans = do transports <- atomically $ newTVar trans bindings <- atomically $ newTVar M.empty - mailbox <- atomically $ newTQueue + mailbox <- atomically $ newMailbox return Endpoint { endpointTransports = transports, endpointBindings = bindings, @@ -195,7 +202,7 @@ Receive the next 'Message' sent to the 'Endpoint', blocking until a message is available. -} receiveMessage :: Endpoint -> IO Message -receiveMessage endpoint = atomically $ readTQueue $ endpointMailbox endpoint +receiveMessage endpoint = atomically $ readMailbox $ endpointMailbox endpoint {-| Wait for a message to be received within the timeout, blocking until either a message @@ -205,6 +212,55 @@ receiveMessageTimeout :: Endpoint -> Int -> IO (Maybe Message) receiveMessageTimeout endpoint delay = do resultOrTimeout <- race (receiveMessage endpoint) (threadDelay delay) + case resultOrTimeout of + Left result -> return $ Just result + Right () -> return Nothing + +{-| +Posts a 'Message' directly to an 'Endpoint', without use of a transport. This +may be useful for applications that prefer to use the 'Endpoint''s 'Mailbox' +as a general queue of ordered messages. +-} +postMessage :: Endpoint -> Message -> IO () +postMessage endpoint message = do + atomically $ writeMailbox (endpointMailbox endpoint) message + +{-| +Select the next available message in the 'Endpoint' 'Mailbox' matching +the supplied test function, or blocking until one is available. This function +differs from 'receiveMessage' in that it supports out of order message reception. +-} +selectMessage :: Endpoint -> (Message -> Maybe v) -> IO v +selectMessage endpoint testFn = do + msg <- atomically $ selectMailbox (endpointMailbox endpoint) testFn + return msg + +{-| +Wait for a message to be selected within the timeout, blocking until either a message +is available or the timeout has occurred. If a message was available, returns @Just message@, +but returns @Nothing@ if no message available before the timeout occurred. Like +'selectMessage', this function enables out of order message reception. +-} +selectMessageTimeout :: Endpoint -> Int -> (Message -> Maybe v) -> IO (Maybe v) +selectMessageTimeout endpoint delay testFn = do + resultOrTimeout <- race (selectMessage endpoint testFn) (threadDelay delay) + case resultOrTimeout of + Left result -> return $ Just result + Right () -> return Nothing + +{-| +Dispatch the next available message in the 'Endpoint' 'Mailbox' matching +the supplied test function, or blocking until one is available. Once a +matching message is found, handle the message with the supplied handler +and return any result obtained. This function differs from 'receiveMessage' +in that it supports out of order message reception. +-} +dispatchMessage :: Endpoint -> (Message -> Maybe v) -> (v -> IO r) -> IO r +dispatchMessage endpoint = handleMailbox (endpointMailbox endpoint) + +dispatchMessageTimeout :: Endpoint -> Int -> (Message -> Maybe v) -> (v -> IO r) -> IO (Maybe r) +dispatchMessageTimeout endpoint delay testFn handleFn = do + resultOrTimeout <- race (dispatchMessage endpoint testFn handleFn) (threadDelay delay) case resultOrTimeout of Left result -> return $ Just result Right () -> return Nothing diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs --- a/src/Network/Transport.hs +++ b/src/Network/Transport.hs @@ -20,21 +20,22 @@ Address, Binding(..), Envelope(..), - Mailbox, - newMailbox, Message, Name, Resolver, resolve, resolverFromList, Scheme, - Transport(..), + Transport(..), + + module Control.Concurrent.Mailbox + ) where -- local imports +import Control.Concurrent.Mailbox -- external imports -import Control.Concurrent.STM import Data.ByteString as B import Data.Serialize @@ -72,15 +73,8 @@ to receive. Typically 'Network.Endpoint.Endpoint's will use the same 'Mailbox' when binding or connecting with a 'Transport'. -} -type Mailbox = TQueue Message {-| -Create a new mailbox. --} -newMailbox :: IO Mailbox -newMailbox = atomically $ newTQueue - -{-| An address is a logical identifier suitable for establishing a connection to another 'Endpoint' over a 'Transport'. It's use (if at all) is specific to the 'Transport' in question. @@ -108,7 +102,7 @@ data Transport = Transport { scheme :: String, handles :: Name -> IO Bool, - bind :: Mailbox -> Name -> IO (Either String Binding), + bind :: Mailbox Message -> Name -> IO (Either String Binding), sendTo :: Name -> Message -> IO (), shutdown :: IO () } 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 @@ -38,7 +38,7 @@ memoryScheme = "mem" data MemoryTransport = MemoryTransport { - boundMailboxes :: TVar (M.Map Name Mailbox) + boundMailboxes :: TVar (M.Map Name (Mailbox Message)) } {-| @@ -58,7 +58,7 @@ shutdown = return () } -memoryBind :: MemoryTransport -> Mailbox -> Name -> IO (Either String Binding) +memoryBind :: MemoryTransport -> Mailbox Message -> Name -> IO (Either String Binding) memoryBind transport mailbox name = do atomically $ modifyTVar (boundMailboxes transport) (\mailboxes -> M.insert name mailbox mailboxes) @@ -75,7 +75,7 @@ memorySendTo transport name msg = do mailboxes <- atomically $ readTVar $ boundMailboxes transport case M.lookup name mailboxes of - Just mailbox -> atomically $ writeTQueue mailbox msg + Just mailbox -> atomically $ writeMailbox mailbox msg Nothing -> return () -- error $ "No mailbox for " ++ name memoryUnbind :: MemoryTransport -> Name -> IO () diff --git a/src/Network/Transport/Sockets.hs b/src/Network/Transport/Sockets.hs --- a/src/Network/Transport/Sockets.hs +++ b/src/Network/Transport/Sockets.hs @@ -19,6 +19,11 @@ Bindings, + newSocketBindings, + SocketBindings, + bindAddress, + unbindAddress, + SocketTransport(..), Connection(..), @@ -35,6 +40,7 @@ sender, socketSendTo, receiver, + receiveSocketBytes, receiveSocketMessage, receiveSocketMessages, SocketSend, @@ -64,8 +70,8 @@ import GHC.Generics -import Network.Simple.TCP (recv) import Network.Socket hiding (recv,socket) +import qualified Network.Socket.ByteString as NSB import System.Log.Logger @@ -75,14 +81,101 @@ _log :: String _log = "transport.sockets" -type Bindings = TVar (M.Map Name Mailbox) +type Bindings = TVar (M.Map Name (Mailbox Message)) +data SocketBinding = SocketBinding { + socketCount :: TVar Int, + socketSocket :: TMVar Socket, + socketListener :: TMVar (Async ()) +} + +type SocketBindings = TVar (M.Map Address SocketBinding) + +newSocketBindings :: IO SocketBindings +newSocketBindings = atomically $ newTVar M.empty + +bindAddress :: SocketBindings -> Address -> IO (Socket,Async ()) -> IO () +bindAddress bindings address factory = do + (count,binding) <- atomically $ do + bmap <- readTVar bindings + bndg <- case M.lookup address bmap of + Nothing -> do + count <- newTVar 1 + listener <- newEmptyTMVar + sock <- newEmptyTMVar + let binding = SocketBinding { + socketCount = count, + socketListener = listener, + socketSocket = sock + } + modifyTVar bindings $ \bs -> M.insert address binding bs + return binding + Just binding -> do + modifyTVar (socketCount binding) $ \c -> c + 1 + return binding + count <- readTVar $ socketCount bndg + return (count,bndg) + if count == 1 + then do + infoM _log $ "Opening binding for " ++ (show address) + (sock,listener) <- factory + infoM _log $ "Opened binding for " ++ (show address) + atomically $ do + putTMVar (socketSocket binding) sock + putTMVar (socketListener binding) listener + return () + else return () + +unbindAddress :: SocketBindings -> Address -> IO () +unbindAddress bindings address = do + (count, maybeBinding) <- atomically $ do + bmap <- readTVar bindings + case M.lookup address bmap of + Nothing -> return (0,Nothing) + Just b -> do + modifyTVar (socketCount b) $ \count -> count - 1 + count <- readTVar (socketCount b) + return (count,Just b) + case maybeBinding of + -- no binding to shutdown; can just return + Nothing -> do + warningM _log $ "No binding for " ++ (show address) ++ "; count is " ++ (show count) + return () + -- we're the last, so close the binding + Just binding -> do + if count == 0 + then do + (sock,listener) <- atomically $ do + sock <- takeTMVar $ socketSocket binding + listener <- takeTMVar $ socketListener binding + return (sock,listener) + infoM _log $ "Closing binding for " ++ (show address) ++ "; count is " ++ (show count) + cancel listener + sClose sock + infoM _log $ "Closed binding for " ++ (show address) + -- now we remove the binding from the map, if it is still there and 0 + -- the expectation here is that only one thread is going to remove it + -- from the map + atomically $ do + bmap <- readTVar bindings + case M.lookup address bmap of + Nothing -> return () + Just b -> do + newCount <- readTVar (socketCount b) + if newCount == 0 + then do + modifyTVar bindings $ \bm -> M.delete address bm + return () + else return () + else return () + return () + data SocketTransport = SocketTransport { socketMessengers :: TVar (M.Map Address Messenger), socketBindings :: Bindings, socketConnection :: Address -> IO Connection, - socketMessenger :: Connection -> Mailbox -> IO Messenger, - socketInbound :: Mailbox, + socketMessenger :: Connection -> Mailbox Message -> IO Messenger, + socketInbound :: Mailbox Message, socketDispatchers :: S.Set (Async ()), socketResolver :: Resolver @@ -109,12 +202,13 @@ when any send / receive action actually completes. -} data Messenger = Messenger { - messengerOut :: Mailbox, - messengerAddress :: Address, - messengerSender :: Async (), - messengerReceiver :: Async (), - messengerConnection :: Connection - } + messengerDone :: TVar Bool, + messengerOut :: Mailbox Message, + messengerAddress :: Address, + messengerSender :: Async (), + messengerReceiver :: Async (), + messengerConnection :: Connection + } data IdentifyMessage = IdentifyMessage Address deriving (Generic) @@ -158,12 +252,14 @@ instance Show Messenger where show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")" -newMessenger :: Connection -> Mailbox -> IO Messenger +newMessenger :: Connection -> Mailbox Message -> IO Messenger newMessenger conn inc = do - out <- newMailbox - sndr <- async $ sender conn out - rcvr <- async $ receiver conn inc + out <- atomically $ newMailbox + done <- atomically $ newTVar False + sndr <- async $ sender conn done out + rcvr <- async $ receiver conn done inc return Messenger { + messengerDone = done, messengerOut = out, messengerAddress = connAddress conn, messengerSender = sndr, @@ -180,14 +276,14 @@ 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 () +deliver msngr message = atomically $ writeMailbox (messengerOut msngr) message + +dispatcher :: TVar (M.Map Name (Mailbox Message)) -> Mailbox Message -> IO () dispatcher bindings mbox = dispatchMessages where dispatchMessages = catchExceptions (do infoM _log $ "Dispatching messages" - env <- atomically $ readTQueue mbox + env <- atomically $ readMailbox mbox dispatchMessage env dispatchMessages) (\e -> do @@ -206,17 +302,17 @@ case maybeDest of Nothing -> return () Just dest -> do - writeTQueue dest msg + writeMailbox dest msg return () -sender :: Connection -> Mailbox -> IO () -sender conn mailbox = sendMessages +sender :: Connection -> TVar Bool -> Mailbox Message -> IO () +sender conn done mailbox = sendMessages where sendMessages = do reconnect catchExceptions (do infoM _log $ "Waiting to send to " ++ (show $ connAddress conn) - msg <- atomically $ readTQueue mailbox + msg <- atomically $ readMailbox mailbox infoM _log $ "Sending message to " ++ (show $ connAddress conn) connected <- atomically $ tryReadTMVar $ connSocket conn case connected of @@ -227,7 +323,10 @@ (\e -> do warningM _log $ "Send error: " ++ (show (e :: SomeException)) disconnect) - sendMessages + isDone <- atomically $ readTVar done + if isDone + then return () + else sendMessages reconnect = do -- TODO need a timeout here, in case connecting always fails connected <- atomically $ tryReadTMVar $ connSocket conn @@ -260,7 +359,7 @@ case found of Nothing -> return False Just mbox -> do - atomically $ writeTQueue mbox msg + atomically $ writeMailbox mbox msg return True remote = do Just address <- resolve (socketResolver transport) name @@ -284,42 +383,62 @@ return () Just msngr -> deliver msngr env -receiver :: Connection -> Mailbox -> IO () -receiver conn mailbox = do +receiver :: Connection -> TVar Bool -> Mailbox Message -> IO () +receiver conn done mailbox = do socket <- atomically $ readTMVar $ connSocket conn - receiveSocketMessages socket (connAddress conn) mailbox + receiveSocketMessages socket done (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))) +receiveSocketMessages :: Socket -> TVar Bool -> Address -> Mailbox Message -> IO () +receiveSocketMessages sock done addr mailbox = do + 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 $ writeMailbox mailbox msg + isDone <- atomically $ readTVar done + if isDone + then return () + else receiveSocketMessages sock done addr mailbox) + (\e -> do + isDone <- atomically $ readTVar done + if isDone + then return () + -- Dropping this message to info, as even well-behaved applications + -- may generate it...even though it is benign + else infoM _log $ "Receive error: " ++ (show (e :: SomeException))) -receiveSocketMessage :: Socket -> IO (Maybe Message) +receiveSocketMessage :: Socket -> IO (Maybe B.ByteString) receiveSocketMessage socket = do - maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes + maybeLen <- receiveSocketBytes 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) + maybeMsg <- receiveSocketBytes socket $ msgLength (decode len) infoM _log $ "Received message" return maybeMsg where msgLength (Right size) = size msgLength (Left err) = error err +receiveSocketBytes :: Socket -> Int -> IO (Maybe B.ByteString) +receiveSocketBytes sock maxBytes = do + bs <- NSB.recv sock maxBytes + if B.null bs + then return Nothing + else return $ Just bs + closeMessenger :: Messenger -> IO () closeMessenger msngr = do + infoM _log $ "Closing mesenger to " ++ (messengerAddress msngr) + atomically $ modifyTVar (messengerDone msngr) (\_ -> True) cancel $ messengerSender msngr cancel $ messengerReceiver msngr connClose $ messengerConnection msngr + infoM _log $ "Closed messenger to " ++ (messengerAddress 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 @@ -39,8 +39,8 @@ import Data.Serialize import qualified Data.Set as S -import Network.Socket (sClose,accept) -import Network.Simple.TCP hiding (accept) +import qualified Network.Socket as NS +import qualified Network.Socket.ByteString as NSB import System.Log.Logger @@ -63,21 +63,22 @@ newTCPTransport resolver = do messengers <- atomically $ newTVar M.empty bindings <- atomically $ newTVar M.empty - inbound <- newMailbox + sockets <- newSocketBindings + inbound <- atomically $ newMailbox dispatch <- async $ dispatcher bindings inbound let transport = SocketTransport { socketMessengers = messengers, socketBindings = bindings, - socketInbound = inbound, socketConnection = newTCPConnection, socketMessenger = newTCPMessenger bindings resolver, + socketInbound = inbound, socketDispatchers = S.fromList [dispatch], socketResolver = resolver } return Transport { scheme = tcpScheme, handles = tcpHandles transport, - bind = tcpBind transport, + bind = tcpBind transport sockets, sendTo = socketSendTo transport, shutdown = tcpShutdown transport } @@ -90,42 +91,50 @@ isJust (Just _) = True isJust _ = False -tcpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding) -tcpBind transport inc name = do - atomically $ modifyTVar (socketBindings transport) $ \bindings -> +tcpBind :: SocketTransport -> SocketBindings -> Mailbox Message -> Name -> IO (Either String Binding) +tcpBind transport sockets inc name = do + atomically $ modifyTVar (socketBindings transport) $ \ bindings -> M.insert name inc bindings Just address <- resolve (socketResolver transport) name - let (_,port) = parseSocketAddress address - listener <- async $ do - infoM _log $ "Binding to address " ++ (show address) - tcpListen address port + bindAddress sockets address $ do + sock <- NS.socket NS.AF_INET NS.Stream NS.defaultProtocol + listener <- do + infoM _log $ "Binding to address " ++ (show address) + tcpListen address sock + return (sock,listener) return $ Right Binding { bindingName = name, - unbind = tcpUnbind listener address + unbind = tcpUnbind address } where - tcpListen address port = - listen HostAny port $ \(socket,_) -> - catchExceptions (do - tcpAccept address socket) + tcpListen address sock = do + catchExceptions (do + let (_,port) = parseSocketAddress address + portnums <- NS.getAddrInfo + (Just NS.defaultHints {NS.addrFlags = [NS.AI_PASSIVE,NS.AI_NUMERICSERV]}) Nothing (Just port) + NS.setSocketOption sock NS.NoDelay 1 + NS.setSocketOption sock NS.ReuseAddr 1 + NS.bind sock (NS.addrAddress $ head portnums) + NS.listen sock 2048) -- TODO think about a configurable backlog (\e -> do - warningM _log $ "Listen error: " ++ (show (e :: SomeException))) - tcpAccept address socket = do - infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show socket) - (client,clientAddress) <- accept socket + warningM _log $ "Listen error on port " ++ address ++ ": " ++ (show (e :: SomeException)) + NS.sClose sock) + async $ tcpAccept address sock + tcpAccept address sock = do + infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show sock) + (client,clientAddress) <- NS.accept sock _ <- async $ tcpDispatch address client clientAddress - tcpAccept address socket + tcpAccept address sock tcpDispatch address client socketAddress = do infoM _log $ "Accepted connection on " ++ (show address) identity <- tcpIdentify client socketAddress case identity of - Nothing -> sClose client + Nothing -> NS.sClose client Just (IdentifyMessage clientAddress) -> do infoM _log $ "Identified " ++ (show clientAddress) - clientSocket <- atomically $ newTMVar client newConn <- newTCPConnection clientAddress - let conn = newConn {connSocket = clientSocket} - msngr <- newMessenger conn (socketInbound transport) + atomically $ putTMVar (connSocket newConn) client + msngr <- newMessenger newConn (socketInbound transport) found <- atomically $ do msngrs <- readTVar $ socketMessengers transport return $ M.lookup clientAddress msngrs @@ -145,9 +154,10 @@ case msg of Left _ -> return Nothing Right message -> return $ Just message - tcpUnbind listener address = do - infoM _log $ "Unbinding from port " ++ (show address) - cancel listener + tcpUnbind address = do + infoM _log $ "Unbinding from TCP port " ++ (show address) + unbindAddress sockets address + infoM _log $ "Unbound from TCP port " ++ (show address) newTCPConnection :: Address -> IO Connection newTCPConnection address = do @@ -157,19 +167,29 @@ connAddress = address, connSocket = sock, connConnect = do - (s,_) <- connectSock host port - return s, - -- connSend = send, + socket <- NS.socket NS.AF_INET NS.Stream NS.defaultProtocol + sockAddrs <- NS.getAddrInfo + (Just NS.defaultHints {NS.addrFlags = [NS.AI_NUMERICSERV]}) (Just host) (Just port) + NS.connect socket $ NS.addrAddress $ head sockAddrs + atomically $ putTMVar sock socket + return socket, connSend = tcpSend address, - connReceive = recv, + connReceive = receiveSocketBytes, connClose = do + infoM _log $ "Closing connection to " ++ address open <- atomically $ tryTakeTMVar sock case open of - Just socket -> sClose socket - Nothing -> return () + Just socket -> do + infoM _log $ "Closing socket " ++ (show socket) ++ " for " ++ address + NS.sClose socket + infoM _log $ "Closed socket " ++ (show socket) ++ " for " ++ address + Nothing -> do + infoM _log $ "No socket to close for " ++ address + return () + infoM _log $ "Connection to " ++ address ++ " closed" } -newTCPMessenger :: Bindings -> Resolver -> Connection -> Mailbox -> IO Messenger +newTCPMessenger :: Bindings -> Resolver -> Connection -> Mailbox Message -> IO Messenger newTCPMessenger bindings resolver conn mailbox = do msngr <- newMessenger conn mailbox identifyAll msngr @@ -185,11 +205,11 @@ Nothing -> return() Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress -tcpSend :: Address -> Socket -> B.ByteString -> IO () +tcpSend :: Address -> NS.Socket -> B.ByteString -> IO () tcpSend addr sock bs = do - send sock $ encode (B.length bs) + NSB.sendAll sock $ encode (B.length bs) infoM _log $ "Length sent" - send sock bs + NSB.sendAll sock bs infoM _log $ "Message sent to" ++ (show addr) diff --git a/src/Network/Transport/UDP.hs b/src/Network/Transport/UDP.hs --- a/src/Network/Transport/UDP.hs +++ b/src/Network/Transport/UDP.hs @@ -64,21 +64,22 @@ newUDPTransport resolver = do messengers <- atomically $ newTVar M.empty bindings <- atomically $ newTVar M.empty - inbound <- newMailbox + sockets <- newSocketBindings + inbound <- atomically $ newMailbox dispatch <- async $ dispatcher bindings inbound let transport = SocketTransport { socketMessengers = messengers, socketBindings = bindings, - socketInbound = inbound, socketConnection = newUDPConnection, socketMessenger = newUDPMessenger, + socketInbound = inbound, socketDispatchers = S.fromList [dispatch], socketResolver = resolver } return Transport { scheme = udpScheme, handles = udpHandles transport, - bind = udpBind transport, + bind = udpBind transport sockets, sendTo = socketSendTo transport, shutdown = udpShutdown transport } @@ -91,30 +92,32 @@ isJust (Just _) = True isJust _ = False -udpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding) -udpBind transport inc name = do +udpBind :: SocketTransport -> SocketBindings -> Mailbox Message -> Name -> IO (Either String Binding) +udpBind transport sockets 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) + bindAddress sockets address $ do + let (_,port) = parseSocketAddress address + addrinfos <- N.getAddrInfo + (Just (N.defaultHints {N.addrFlags = [N.AI_PASSIVE,N.AI_NUMERICSERV]})) + 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 (sock,rcvr) return $ Right Binding { bindingName = name, unbind = do - cancel rcvr - infoM _log $ "Closing UDP socket bound on " ++ (show port) - N.sClose sock - } + infoM _log $ "Unbinding from UDP port " ++ (show address) + unbindAddress sockets address + infoM _log $ "Unbound from UDP port " ++ (show address) + } newUDPConnection :: Address -> IO Connection newUDPConnection address = do @@ -137,24 +140,26 @@ return () } -newUDPMessenger :: Connection -> Mailbox -> IO Messenger +newUDPMessenger :: Connection -> Mailbox Message -> 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))) +udpReceiveSocketMessages :: N.Socket -> Address -> Mailbox Message -> 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 $ writeMailbox mailbox msg + udpReceiveSocketMessages sock addr mailbox) + (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException))) where udpReceiveSocketMessage = do maybeMsg <- udpRecvFrom sock 512 diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -15,6 +15,7 @@ import Test.Framework.Providers.HUnit -- Test modules +import qualified TestMailbox as MB import qualified TestMemory as M import qualified TestTCP as T import qualified TestUDP as U @@ -44,6 +45,7 @@ testCase "hunit" (assertBool "HUnit assertion of truth is false" True), testCase "endpoints" testEndpoint ] + ++ MB.tests ++ M.tests ++ T.tests ++ U.tests @@ -54,4 +56,3 @@ testEndpoint = do _ <- newEndpoint [] return () -