network-transport 0.3.0.1 → 0.5.9
raw patch · 6 files changed
Files
- ChangeLog +91/−0
- Setup.hs +0/−2
- network-transport.cabal +42/−23
- src/Network/Transport.hs +59/−44
- src/Network/Transport/Internal.hs +99/−53
- src/Network/Transport/Util.hs +15/−20
+ ChangeLog view
@@ -0,0 +1,91 @@+2025-02-04 Laurent P. René de Cotret <laurent.decotret@outlook.com> 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 <laurent.decotret@outlook.com> 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 <dsd@caltech.edu> 0.5.7++* Bump bytestring and deepseq versions to build with GHC 9.8.++2022-08-30 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.6++* Fix extension fields in the cabal file (#41).++2022-08-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.5++* Relax dependency bounds to build with ghc9 (#40).++2019-05-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.4++* Fix documentation typo (#39).++2019-05-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.3++* Relax upper bound of hashable.++2017-07-25 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.5.1++* Add {encode|decode}{Word|Enum|Num}{32|16}.+* Removed {encode|decode}Int{32|16}++2016-01-28 Facundo Domínguez <facundo.dominguez@tweag.io> 0.4.4.0++* Add compatibility with ghc-8.++2016-01-28 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.4.2.0++* Add NFData instance for EndPointAddress.+* Relax dependency bounds.++2014-12-09 Tim Watson <watson.timothy@gmail.com> 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 <watson.timothy@gmail.com> 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 <edsko@well-typed.com> 0.3.0.1++* Relax bounds on Binary++2012-10-03 Edsko de Vries <edsko@well-typed.com> 0.3.0++* Clarify disconnection+* Require that 'connect' be "as asynchronous as possible"+* Added strictness annotations++2012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.0.2++* Base 4.6 compatible test suites+* Relax package constraints for bytestring++2012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.0.1++* Hide catch only for base < 4.6++2012-07-07 Edsko de Vries <edsko@well-typed.com> 0.2.0++* Initial release.
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
network-transport.cabal view
@@ -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 (<http://www.olcf.ornl.gov/center-projects/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
src/Network/Transport.hs view
@@ -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
src/Network/Transport/Internal.hs view
@@ -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
src/Network/Transport/Util.hs view
@@ -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