courier 0.1.0.7 → 0.1.0.8
raw patch · 10 files changed
+638/−146 lines, 10 files
Files
- changes.md +23/−0
- courier.cabal +10/−9
- src/Control/Concurrent/Mailbox.hs +273/−0
- src/Network/Endpoints.hs +60/−4
- src/Network/Transport.hs +6/−12
- src/Network/Transport/Memory.hs +3/−3
- src/Network/Transport/Sockets.hs +162/−43
- src/Network/Transport/TCP.hs +59/−39
- src/Network/Transport/UDP.hs +40/−35
- tests/Tests.hs +2/−1
changes.md view
@@ -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
courier.cabal view
@@ -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
+ src/Control/Concurrent/Mailbox.hs view
@@ -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
src/Network/Endpoints.hs view
@@ -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
src/Network/Transport.hs view
@@ -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 () }
src/Network/Transport/Memory.hs view
@@ -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 ()
src/Network/Transport/Sockets.hs view
@@ -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)
src/Network/Transport/TCP.hs view
@@ -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)
src/Network/Transport/UDP.hs view
@@ -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
tests/Tests.hs view
@@ -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 ()-