diff --git a/ChangeLog b/ChangeLog new file mode 100644 --- /dev/null +++ b/ChangeLog @@ -0,0 +1,91 @@ +2025-02-04 Laurent P. René de Cotret 0.5.9 + +* Prevent asynchronous exceptions from `Network.Transport.Util.spawn` (#42). +* Addressed some compilation warnings (#467) + +2024-09-03 Laurent P. René de Cotret 0.5.8 + +* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1 +* Updated links to point to Distributed Haskell monorepo + +2024-03-25 David Simmons-Duffin 0.5.7 + +* Bump bytestring and deepseq versions to build with GHC 9.8. + +2022-08-30 Facundo Domínguez 0.5.6 + +* Fix extension fields in the cabal file (#41). + +2022-08-09 Facundo Domínguez 0.5.5 + +* Relax dependency bounds to build with ghc9 (#40). + +2019-05-12 Facundo Domínguez 0.5.4 + +* Fix documentation typo (#39). + +2019-05-12 Facundo Domínguez 0.5.3 + +* Relax upper bound of hashable. + +2017-07-25 Facundo Domínguez 0.5.2 + +* prependLength checks for overflow (5608f0f) +* Drop inlinePerformIO for unsafeDupablePerformIO (18bf80c) +* Have travis build n-t even with no tests (7ffe43e) + +2017-02-23 Facundo Domínguez 0.5.1 + +* Add {encode|decode}{Word|Enum|Num}{32|16}. +* Removed {encode|decode}Int{32|16} + +2016-01-28 Facundo Domínguez 0.4.4.0 + +* Add compatibility with ghc-8. + +2016-01-28 Facundo Domínguez 0.4.3.0 + +* Derive Binary instances for missing types. +* Use auto-derive for Reliability as Binary instance. +* Stop testing with ghc-7.4 and build with ghc-7.10. + +2015-06-15 Facundo Domínguez 0.4.2.0 + +* Add NFData instance for EndPointAddress. +* Relax dependency bounds. + +2014-12-09 Tim Watson 0.4.1.0 + +* foreigns htonl, ntohl, htons, ntohs are imported from ws2_32 on windows +* Created Data instance for EndPointAddress (thanks Andrew Rademacher) + +2014-05-30 Tim Watson 0.4.0.0 + +* Fix build for GHC 7.4 - thanks mboes! +* Allow transformers above v5 +* Bump binary version to include 0.7.* +* Binary instance for 'Reliability' - thanks mboes! +* Hashable instance for 'EndPointAddress' + +2012-11-22 Edsko de Vries 0.3.0.1 + +* Relax bounds on Binary + +2012-10-03 Edsko de Vries 0.3.0 + +* Clarify disconnection +* Require that 'connect' be "as asynchronous as possible" +* Added strictness annotations + +2012-07-16 Edsko de Vries 0.2.0.2 + +* Base 4.6 compatible test suites +* Relax package constraints for bytestring + +2012-07-16 Edsko de Vries 0.2.0.1 + +* Hide catch only for base < 4.6 + +2012-07-07 Edsko de Vries 0.2.0 + +* Initial release. diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/network-transport.cabal b/network-transport.cabal --- a/network-transport.cabal +++ b/network-transport.cabal @@ -1,16 +1,16 @@ +cabal-version: 3.0 Name: network-transport -Version: 0.3.0.1 -Cabal-Version: >=1.6 +Version: 0.5.9 Build-Type: Simple -License: BSD3 +License: BSD-3-Clause License-File: LICENSE Copyright: Well-Typed LLP Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: edsko@well-typed.com, duncan@well-typed.com +maintainer: The Distributed Haskell team Stability: experimental -Homepage: http://github.com/haskell-distributed/distributed-process -Bug-Reports: mailto:edsko@well-typed.com -Synopsis: Network abstraction layer +Homepage: https://haskell-distributed.github.io +Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues +Synopsis: Network abstraction layer Description: "Network.Transport" is a Network Abstraction Layer which provides the following high-level concepts: . @@ -23,7 +23,7 @@ using the 'EndPointAddress' of the remote end. . * The 'EndPointAddress' can be serialised and sent over the - network, where as 'EndPoint's and connections cannot. + network, whereas 'EndPoint's and connections cannot. . * Connections between 'EndPoint's are unidirectional and lightweight. . @@ -35,12 +35,12 @@ . * In addition to incoming messages, 'EndPoint's are notified of other 'Event's such as new connections or broken connections. - . + . This design was heavily influenced by the design of the Common Communication Interface (). Important design goals are: - . + . * Connections should be lightweight: it should be no problem to create thousands of connections between endpoints. . @@ -50,32 +50,51 @@ * Error handling is "abstract": errors that originate from implementation specific problems (such as "no more sockets" in the TCP implementation) get mapped to generic errors - ("insufficient resources") at the Transport level. + ("insufficient resources") at the Transport level. . This package provides the generic interface only; you will probably also want to install at least one transport implementation (network-transport-*). -Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 -Category: Network +tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1 +Category: Network +extra-doc-files: ChangeLog -Source-Repository head +source-repository head Type: git Location: https://github.com/haskell-distributed/distributed-process - SubDir: network-transport + SubDir: packages/network-transport +common warnings + ghc-options: -Wall + -Wcompat + -Widentities + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wredundant-constraints + -fhide-source-paths + -Wpartial-fields + -Wunused-packages + Library - Build-Depends: base >= 4.3 && < 5, - binary >= 0.5 && < 0.7, - bytestring >= 0.9 && < 0.11, - transformers >= 0.2 && < 0.4 - Exposed-Modules: Network.Transport, + import: warnings + Build-Depends: base >= 4.14 && < 5, + binary >= 0.8 && < 0.9, + bytestring >= 0.10 && < 0.13, + hashable >= 1.2.0.5 && < 1.6, + transformers >= 0.2 && < 0.7, + deepseq >= 1.0 && < 1.7 + Exposed-Modules: Network.Transport Network.Transport.Util Network.Transport.Internal - Extensions: ForeignFunctionInterface, - RankNTypes, + Other-Extensions: ForeignFunctionInterface + Default-Extensions: + RankNTypes, ScopedTypeVariables, DeriveDataTypeable, GeneralizedNewtypeDeriving, CPP - GHC-Options: -Wall -fno-warn-unused-do-bind + GHC-Options: -fno-warn-unused-do-bind HS-Source-Dirs: src + Default-Language: Haskell2010 + if os(win32) + extra-libraries: ws2_32 diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs --- a/src/Network/Transport.hs +++ b/src/Network/Transport.hs @@ -1,5 +1,6 @@ --- | Network Transport -module Network.Transport +{-# LANGUAGE DeriveGeneric #-} +-- | Network Transport +module Network.Transport ( -- * Types Transport(..) , EndPoint(..) @@ -26,11 +27,14 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS (copy) import qualified Data.ByteString.Char8 as BSC (unpack) +import Control.DeepSeq (NFData(rnf)) import Control.Exception (Exception) -import Control.Applicative ((<$>)) import Data.Typeable (Typeable) -import Data.Binary (Binary(get, put)) +import Data.Binary (Binary(..)) +import Data.Hashable import Data.Word (Word64) +import Data.Data (Data) +import GHC.Generics (Generic) -------------------------------------------------------------------------------- -- Main API -- @@ -41,8 +45,8 @@ data Transport = Transport { -- | Create a new end point (heavyweight operation) newEndPoint :: IO (Either (TransportError NewEndPointErrorCode) EndPoint) - -- | Shutdown the transport completely - , closeTransport :: IO () + -- | Shutdown the transport completely + , closeTransport :: IO () } -- | Network endpoint. @@ -50,8 +54,8 @@ -- | Endpoints have a single shared receive queue. receive :: IO Event -- | EndPointAddress of the endpoint. - , address :: EndPointAddress - -- | Create a new lightweight connection. + , address :: EndPointAddress + -- | Create a new lightweight connection. -- -- 'connect' should be as asynchronous as possible; for instance, in -- Transport implementations based on some heavy-weight underlying network @@ -64,12 +68,12 @@ , resolveMulticastGroup :: MulticastAddress -> IO (Either (TransportError ResolveMulticastGroupErrorCode) MulticastGroup) -- | Close the endpoint , closeEndPoint :: IO () - } + } -- | Lightweight connection to an endpoint. data Connection = Connection { -- | Send a message on this connection. - -- + -- -- 'send' provides vectored I/O, and allows multiple data segments to be -- sent using a single call (cf. 'Network.Socket.ByteString.sendMany'). -- Note that this segment structure is entirely unrelated to the segment @@ -80,7 +84,7 @@ } -- | Event on an endpoint. -data Event = +data Event = -- | Received a message Received {-# UNPACK #-} !ConnectionId [ByteString] -- | Connection closed @@ -88,33 +92,36 @@ -- | Connection opened -- -- 'ConnectionId's need not be allocated contiguously. - | ConnectionOpened {-# UNPACK #-} !ConnectionId Reliability EndPointAddress + | ConnectionOpened {-# UNPACK #-} !ConnectionId Reliability EndPointAddress -- | Received multicast | ReceivedMulticast MulticastAddress [ByteString] -- | The endpoint got closed (manually, by a call to closeEndPoint or closeTransport) | EndPointClosed - -- | An error occurred - | ErrorEvent (TransportError EventErrorCode) - deriving (Show, Eq) + -- | An error occurred + | ErrorEvent (TransportError EventErrorCode) + deriving (Show, Eq, Generic) +instance Binary Event + -- | Connection data ConnectHintsIDs enable receivers to distinguish one connection from another. -type ConnectionId = Word64 +type ConnectionId = Word64 -- | Reliability guarantees of a connection. -data Reliability = - ReliableOrdered - | ReliableUnordered +data Reliability = + ReliableOrdered + | ReliableUnordered | Unreliable - deriving (Show, Eq) + deriving (Show, Eq, Typeable, Generic) +instance Binary Reliability -- | Multicast group. data MulticastGroup = MulticastGroup { - -- | EndPointAddress of the multicast group. + -- | EndPointAddress of the multicast group. multicastAddress :: MulticastAddress -- | Delete the multicast group completely. , deleteMulticastGroup :: IO () -- | Maximum message size that we can send to this group. - , maxMsgSize :: Maybe Int + , maxMsgSize :: Maybe Int -- | Send a message to the group. , multicastSend :: [ByteString] -> IO () -- | Subscribe to the given multicast group (to start receiving messages from the group). @@ -127,7 +134,7 @@ -- | EndPointAddress of an endpoint. newtype EndPointAddress = EndPointAddress { endPointAddressToByteString :: ByteString } - deriving (Eq, Ord, Typeable) + deriving (Eq, Ord, Typeable, Data, Hashable) instance Binary EndPointAddress where put = put . endPointAddressToByteString @@ -136,10 +143,14 @@ instance Show EndPointAddress where show = BSC.unpack . endPointAddressToByteString +instance NFData EndPointAddress where rnf x = x `seq` () + -- | EndPointAddress of a multicast group. newtype MulticastAddress = MulticastAddress { multicastAddressToByteString :: ByteString } - deriving (Eq, Ord) + deriving (Eq, Ord, Generic) +instance Binary MulticastAddress + instance Show MulticastAddress where show = show . multicastAddressToByteString @@ -172,13 +183,15 @@ -------------------------------------------------------------------------------- -- | Errors returned by Network.Transport API functions consist of an error --- code and a human readable description of the problem +-- code and a human readable description of the problem data TransportError error = TransportError error String - deriving (Show, Typeable) + deriving (Show, Typeable, Generic) +instance (Binary error) => Binary (TransportError error) + -- | Although the functions in the transport API never throw TransportErrors -- (but return them explicitly), application code may want to turn these into --- exceptions. +-- exceptions. instance (Typeable err, Show err) => Exception (TransportError err) -- | When comparing errors we ignore the human-readable strings @@ -190,19 +203,19 @@ -- | Not enough resources NewEndPointInsufficientResources -- | Failed for some other reason - | NewEndPointFailed + | NewEndPointFailed deriving (Show, Typeable, Eq) --- | Connection failure -data ConnectErrorCode = - -- | Could not resolve the address +-- | Connection failure +data ConnectErrorCode = + -- | Could not resolve the address ConnectNotFound -- | Insufficient resources (for instance, no more sockets available) - | ConnectInsufficientResources + | ConnectInsufficientResources -- | Timeout | ConnectTimeout -- | Failed for other reasons (including syntax error) - | ConnectFailed + | ConnectFailed deriving (Show, Typeable, Eq) -- | Failure during the creation of a new multicast group @@ -221,7 +234,7 @@ ResolveMulticastGroupNotFound -- | Failed for some other reason (including syntax error) | ResolveMulticastGroupFailed - -- | Not all transport implementations support multicast + -- | Not all transport implementations support multicast | ResolveMulticastGroupUnsupported deriving (Show, Typeable, Eq) @@ -230,12 +243,12 @@ -- | Connection was closed SendClosed -- | Send failed for some other reason - | SendFailed + | SendFailed deriving (Show, Typeable, Eq) -- | Error codes used when reporting errors to endpoints (through receive) -data EventErrorCode = - -- | Failure of the entire endpoint +data EventErrorCode = + -- | Failure of the entire endpoint EventEndPointFailed -- | Transport-wide fatal error | EventTransportFailed @@ -247,27 +260,29 @@ -- both directions, must now be considered to have failed; they fail as a -- "bundle" of connections, with only a single "bundle" of connections per -- endpoint at any point in time. - -- + -- -- That is, suppose there are multiple connections in either direction -- between endpoints A and B, and A receives a notification that it has -- lost contact with B. Then A must not be able to send any further - -- messages to B on existing connections. + -- messages to B on existing connections. -- -- Although B may not realize /immediately/ that its connection to A has -- been broken, messages sent by B on existing connections should not be -- delivered, and B must eventually get an EventConnectionLost message, - -- too. + -- too. -- -- Moreover, this event must be posted before A has successfully -- reconnected (in other words, if B notices a reconnection attempt from A, -- it must post the EventConnectionLost before acknowledging the connection -- from A) so that B will not receive events about new connections or - -- incoming messages from A without realizing that it got disconnected. - -- + -- incoming messages from A without realizing that it got disconnected. + -- -- If B attempts to establish another connection to A before it realized -- that it got disconnected from A then it's okay for this connection -- attempt to fail, and the EventConnectionLost to be posted at that point, -- or for the EventConnectionLost to be posted and for the new connection -- to be considered the first connection of the "new bundle". - | EventConnectionLost EndPointAddress - deriving (Show, Typeable, Eq) + | EventConnectionLost EndPointAddress + deriving (Show, Typeable, Eq, Generic) + +instance Binary EventErrorCode diff --git a/src/Network/Transport/Internal.hs b/src/Network/Transport/Internal.hs --- a/src/Network/Transport/Internal.hs +++ b/src/Network/Transport/Internal.hs @@ -1,10 +1,15 @@ +{-# LANGUAGE CPP #-} -- | Internal functions -module Network.Transport.Internal +module Network.Transport.Internal ( -- * Encoders/decoders - encodeInt32 - , decodeInt32 - , encodeInt16 - , decodeInt16 + encodeWord32 + , decodeWord32 + , encodeEnum32 + , decodeNum32 + , encodeWord16 + , decodeWord16 + , encodeEnum16 + , decodeNum16 , prependLength -- * Miscellaneous abstractions , mapIOException @@ -19,22 +24,18 @@ , tlog ) where -#if ! MIN_VERSION_base(4,6,0) -import Prelude hiding (catch) -#endif - import Foreign.Storable (pokeByteOff, peekByteOff) -import Foreign.C (CInt(..), CShort(..)) import Foreign.ForeignPtr (withForeignPtr) import Data.ByteString (ByteString) +import Data.List(foldl') import qualified Data.ByteString as BS (length) -import qualified Data.ByteString.Internal as BSI +import qualified Data.ByteString.Internal as BSI ( unsafeCreate , toForeignPtr - , inlinePerformIO ) +import Data.Word (Word32, Word16) import Control.Monad.IO.Class (MonadIO, liftIO) -import Control.Exception +import Control.Exception ( IOException , SomeException , AsyncException @@ -48,51 +49,96 @@ import Control.Concurrent (ThreadId, forkIO) import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar) import GHC.IO (unsafeUnmask) +import System.IO.Unsafe (unsafeDupablePerformIO) import System.Timeout (timeout) --import Control.Concurrent (myThreadId) -foreign import ccall unsafe "htonl" htonl :: CInt -> CInt -foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt -foreign import ccall unsafe "htons" htons :: CShort -> CShort -foreign import ccall unsafe "ntohs" ntohs :: CShort -> CShort +#ifdef mingw32_HOST_OS --- | Serialize 32-bit to network byte order -encodeInt32 :: Enum a => a -> ByteString -encodeInt32 i32 = +foreign import stdcall unsafe "htonl" htonl :: Word32 -> Word32 +foreign import stdcall unsafe "ntohl" ntohl :: Word32 -> Word32 +foreign import stdcall unsafe "htons" htons :: Word16 -> Word16 +foreign import stdcall unsafe "ntohs" ntohs :: Word16 -> Word16 + +#else + +foreign import ccall unsafe "htonl" htonl :: Word32 -> Word32 +foreign import ccall unsafe "ntohl" ntohl :: Word32 -> Word32 +foreign import ccall unsafe "htons" htons :: Word16 -> Word16 +foreign import ccall unsafe "ntohs" ntohs :: Word16 -> Word16 + +#endif + +-- | Serialize 32-bit to network byte order +encodeWord32 :: Word32 -> ByteString +encodeWord32 w32 = BSI.unsafeCreate 4 $ \p -> - pokeByteOff p 0 (htonl . fromIntegral . fromEnum $ i32) + pokeByteOff p 0 (htonl w32) --- | Deserialize 32-bit from network byte order --- Throws an IO exception if this is not a valid integer. -decodeInt32 :: Num a => ByteString -> a -decodeInt32 bs - | BS.length bs /= 4 = throw $ userError "decodeInt32: Invalid length" - | otherwise = BSI.inlinePerformIO $ do - let (fp, offset, _) = BSI.toForeignPtr bs - withForeignPtr fp $ \p -> do - w32 <- peekByteOff p offset - return (fromIntegral . ntohl $ w32) +-- | Deserialize 32-bit from network byte order +-- Throws an IO exception if this is not exactly 32 bits. +decodeWord32 :: ByteString -> Word32 +decodeWord32 bs + | BS.length bs /= 4 = throw $ userError "decodeWord32: not 4 bytes" + | otherwise = unsafeDupablePerformIO $ do + let (fp, offset, _) = BSI.toForeignPtr bs + withForeignPtr fp $ \p -> ntohl <$> peekByteOff p offset --- | Serialize 16-bit to network byte order -encodeInt16 :: Enum a => a -> ByteString -encodeInt16 i16 = +-- | Serialize 16-bit to network byte order +encodeWord16 :: Word16 -> ByteString +encodeWord16 w16 = BSI.unsafeCreate 2 $ \p -> - pokeByteOff p 0 (htons . fromIntegral . fromEnum $ i16) + pokeByteOff p 0 (htons w16) --- | Deserialize 16-bit from network byte order --- Throws an IO exception if this is not a valid integer -decodeInt16 :: Num a => ByteString -> a -decodeInt16 bs - | BS.length bs /= 2 = throw $ userError "decodeInt16: Invalid length" - | otherwise = BSI.inlinePerformIO $ do - let (fp, offset, _) = BSI.toForeignPtr bs - withForeignPtr fp $ \p -> do - w16 <- peekByteOff p offset - return (fromIntegral . ntohs $ w16) +-- | Deserialize 16-bit from network byte order +-- Throws an IO exception if this is not exactly 16 bits. +decodeWord16 :: ByteString -> Word16 +decodeWord16 bs + | BS.length bs /= 2 = throw $ userError "decodeWord16: not 2 bytes" + | otherwise = unsafeDupablePerformIO $ do + let (fp, offset, _) = BSI.toForeignPtr bs + withForeignPtr fp $ \p -> ntohs <$> peekByteOff p offset +-- | Encode an Enum in 32 bits by encoding its signed Int equivalent (beware +-- of truncation, an Enum may contain more than 2^32 points). +encodeEnum32 :: Enum a => a -> ByteString +encodeEnum32 = encodeWord32 . fromIntegral . fromEnum + +-- | Decode any Num type from 32 bits by using fromIntegral to convert from +-- a Word32. +decodeNum32 :: Num a => ByteString -> a +decodeNum32 = fromIntegral . decodeWord32 + +-- | Encode an Enum in 16 bits by encoding its signed Int equivalent (beware +-- of truncation, an Enum may contain more than 2^16 points). +encodeEnum16 :: Enum a => a -> ByteString +encodeEnum16 = encodeWord16 . fromIntegral . fromEnum + +-- | Decode any Num type from 16 bits by using fromIntegral to convert from +-- a Word16. +decodeNum16 :: Num a => ByteString -> a +decodeNum16 = fromIntegral . decodeWord16 + -- | Prepend a list of bytestrings with their total length +-- Will be an exception in case of overflow: the sum of the lengths of +-- the ByteStrings overflows Int, or that sum overflows Word32. prependLength :: [ByteString] -> [ByteString] -prependLength bss = encodeInt32 (sum . map BS.length $ bss) : bss +prependLength bss = case word32Length of + Nothing -> overflow + Just w32 -> encodeWord32 w32 : bss + where + intLength :: Int + intLength = foldl' safeAdd 0 . map BS.length $ bss + word32Length :: Maybe Word32 + word32Length = tryToEnum intLength + -- Non-negative integer addition with overflow check. + safeAdd :: Int -> Int -> Int + safeAdd i j + | r >= 0 = r + | otherwise = overflow + where + r = i + j + overflow = throw $ userError "prependLength: input is too long (overflow)" -- | Translate exceptions that arise in IO computations mapIOException :: Exception e => (IOException -> e) -> IO a -> IO a @@ -120,17 +166,17 @@ forkIOWithUnmask io = forkIO (io unsafeUnmask) -- | Safe version of 'toEnum' -tryToEnum :: (Enum a, Bounded a) => Int -> Maybe a +tryToEnum :: (Enum a, Bounded a) => Int -> Maybe a tryToEnum = go minBound maxBound where go :: Enum b => b -> b -> Int -> Maybe b - go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing + go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing -- | If the timeout value is not Nothing, wrap the given computation with a -- timeout and it if times out throw the specified exception. Identity -- otherwise. timeoutMaybe :: Exception e => Maybe Int -> e -> IO a -> IO a -timeoutMaybe Nothing _ f = f +timeoutMaybe Nothing _ f = f timeoutMaybe (Just n) e f = do ma <- timeout n f case ma of @@ -138,19 +184,19 @@ Just a -> return a -- | @asyncWhenCancelled g f@ runs f in a separate thread and waits for it --- to complete. If f throws an exception we catch it and rethrow it in the +-- to complete. If f throws an exception we catch it and rethrow it in the -- current thread. If the current thread is interrupted before f completes, -- we run the specified clean up handler (if f throws an exception we assume -- that no cleanup is necessary). asyncWhenCancelled :: forall a. (a -> IO ()) -> IO a -> IO a asyncWhenCancelled g f = mask_ $ do mvar <- newEmptyMVar - forkIO $ try f >>= putMVar mvar + forkIO $ try f >>= putMVar mvar -- takeMVar is interruptible (even inside a mask_) catch (takeMVar mvar) (exceptionHandler mvar) >>= either throwIO return where - exceptionHandler :: MVar (Either SomeException a) - -> AsyncException + exceptionHandler :: MVar (Either SomeException a) + -> AsyncException -> IO (Either SomeException a) exceptionHandler mvar ex = do forkIO $ takeMVar mvar >>= either (const $ return ()) g diff --git a/src/Network/Transport/Util.hs b/src/Network/Transport/Util.hs --- a/src/Network/Transport/Util.hs +++ b/src/Network/Transport/Util.hs @@ -1,9 +1,9 @@ --- | Utility functions --- +-- | Utility functions +-- -- Note: this module is bound to change even more than the rest of the API :) module Network.Transport.Util (spawn) where -import Network.Transport +import Network.Transport ( Transport , EndPoint(..) , EndPointAddress @@ -11,23 +11,18 @@ ) import Control.Exception (throwIO) import Control.Concurrent (forkIO) -import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) --- | Fork a new thread, create a new end point on that thread, and run the specified IO operation on that thread. --- +-- | Create a new end point, fork a new thread, and run the specified IO operation on that thread. +-- -- Returns the address of the new end point. -spawn :: Transport -> (EndPoint -> IO ()) -> IO EndPointAddress +spawn :: Transport -> (EndPoint -> IO ()) -> IO EndPointAddress spawn transport proc = do - addrMVar <- newEmptyMVar - forkIO $ do - mEndPoint <- newEndPoint transport - case mEndPoint of - Left err -> - putMVar addrMVar (Left err) - Right endPoint -> do - putMVar addrMVar (Right (address endPoint)) - proc endPoint - mAddr <- takeMVar addrMVar - case mAddr of - Left err -> throwIO err - Right addr -> return addr + -- `newEndPoint` used to be done in a separate thread, in case it was slow. + -- However, in this case, care must be taken to appropriately handle asynchronous exceptions. + -- Instead, for reliability, we now create the new endpoint in this thread. + mEndPoint <- newEndPoint transport + case mEndPoint of + Left err -> throwIO err + Right endPoint -> do + forkIO $ proc endPoint + return $ address endPoint