network-unexceptional 0.1.2.0 → 0.1.3.0
raw patch · 6 files changed
+249/−19 lines, 6 filesdep +stmdep ~basedep ~posix-api
Dependencies added: stm
Dependency ranges changed: base, posix-api
Files
- CHANGELOG.md +7/−0
- network-unexceptional.cabal +3/−2
- src/Network/Unexceptional.hs +103/−4
- src/Network/Unexceptional/Bytes.hs +77/−11
- src/Network/Unexceptional/Chunks.hs +9/−0
- src/Network/Unexceptional/MutableBytes.hs +50/−2
CHANGELOG.md view
@@ -1,5 +1,12 @@ # Revision history for network-unexceptional +## 0.1.3.0 -- 2023-08-30++* Add interruptible send and receive functions to give users control over+ when to give to abandon communicating.+* Add these functions to `Network.Unexceptional`: `connect`,+ `connectInterruptible`, `socket`.+ ## 0.1.1.0 -- 2023-08-14 * Add Network.Unexceptional module with `accept_`.
network-unexceptional.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: network-unexceptional-version: 0.1.2.0+version: 0.1.3.0 synopsis: Network functions that do not throw exceptions description: Functions compatible with the Socket type from the network library that@@ -27,12 +27,13 @@ Network.Unexceptional.Types build-depends: , base >=4.16.3 && <5- , posix-api >=0.6.1+ , posix-api >=0.7 , error-codes >=0.1.1 , byteslice >=0.2.8 , network >=3.1 , primitive >=0.8 , primitive-addr >=0.1.0.2 , bytestring >=0.11.4+ , stm >=2.5.1 hs-source-dirs: src default-language: Haskell2010
src/Network/Unexceptional.hs view
@@ -6,15 +6,30 @@ module Network.Unexceptional ( accept_+ , socket+ , connect+ , connectInterruptible ) where -import Network.Socket (Socket,mkSocket,withFdSocket)-import Foreign.C.Error (Errno)-import GHC.Conc (threadWaitRead)-import Foreign.C.Error.Pattern (pattern EWOULDBLOCK,pattern EAGAIN)+import Control.Exception (mask_)+import Control.Monad ((<=<))+import Control.Applicative ((<|>))+import Data.Functor (($>))+import Network.Socket (Socket,SockAddr,mkSocket,withFdSocket,SocketOption(SoError),getSocketOption)+import Network.Socket.Address (SocketAddress,pokeSocketAddress,sizeOfSocketAddress)+import Foreign.Marshal.Alloc (allocaBytes)+import Foreign.C.Error (Errno(Errno))+import GHC.Conc (threadWaitRead,threadWaitWrite,threadWaitWriteSTM)+import Foreign.C.Error.Pattern (pattern EWOULDBLOCK,pattern EAGAIN,pattern EINPROGRESS,pattern EINTR) import System.Posix.Types (Fd(Fd))+import Foreign.Ptr (castPtr,nullPtr)+import GHC.Exts (Ptr)+import Control.Concurrent.STM (STM,TVar) +import qualified Control.Concurrent.STM as STM import qualified Linux.Socket as X+import qualified Posix.Socket as X+import qualified Network.Socket as N -- | Accept a connection. See the documentation in @network@ for @accept@. --@@ -31,3 +46,87 @@ Right (Fd fd) -> fmap Right (mkSocket fd) acceptLoop +-- | Connect to a socket address. See the documentation in @network@+-- for @connect@.+connect :: Socket -> SockAddr -> IO (Either Errno ())+connect s sa = withSocketAddress sa $ \p_sa sz -> withFdSocket s $ \fd ->+ let loop = do+ r <- X.uninterruptibleConnectPtr (Fd fd) p_sa sz+ case r of+ Right _ -> pure (Right ())+ Left err -> case err of+ EINTR -> loop+ EINPROGRESS -> do+ threadWaitWrite (Fd fd)+ errB <- getSocketOption s SoError+ case errB of+ 0 -> pure (Right ())+ _ -> pure (Left (Errno (fromIntegral errB)))+ _ -> pure (Left err)+ in loop++-- | Variant of 'connect' that can be interrupted by setting the interrupt+-- variable to @True@. If interrupted in this way, this function returns+-- @EAGAIN@. For example, to attempt to connect for no more than 1 second:+--+-- > interrupt <- Control.Concurrent.STM.registerDelay 1_000_000+-- > connectInterruptible interrupt sock sockAddr+connectInterruptible :: TVar Bool -> Socket -> SockAddr -> IO (Either Errno ())+connectInterruptible !interrupt s sa = withSocketAddress sa $ \p_sa sz -> withFdSocket s $ \fd ->+ let loop = do+ r <- X.uninterruptibleConnectPtr (Fd fd) p_sa sz+ case r of+ Right _ -> pure (Right ())+ Left err -> case err of+ EINTR -> loop+ EINPROGRESS -> waitUntilWriteable interrupt (Fd fd) >>= \case+ Interrupted -> pure (Left EAGAIN)+ Ready -> do+ errB <- getSocketOption s SoError+ case errB of+ 0 -> pure (Right ())+ _ -> pure (Left (Errno (fromIntegral errB)))+ _ -> pure (Left err)+ in loop++-- Copied this from the network library. TODO: See if network can+-- just export this.+withSocketAddress :: SocketAddress sa => sa -> (Ptr sa -> Int -> IO a) -> IO a+withSocketAddress addr f = do+ let sz = sizeOfSocketAddress addr+ if sz == 0+ then f nullPtr 0+ else allocaBytes sz $ \p -> pokeSocketAddress p addr >> f (castPtr p) sz++data Outcome = Ready | Interrupted++checkFinished :: TVar Bool -> STM ()+checkFinished = STM.check <=< STM.readTVar++waitUntilWriteable :: TVar Bool -> Fd -> IO Outcome+waitUntilWriteable !interrupt !fd = do+ (isReadyAction,deregister) <- threadWaitWriteSTM fd+ outcome <- STM.atomically $ (isReadyAction $> Ready) <|> (checkFinished interrupt $> Interrupted)+ deregister+ pure outcome++-- | Create a socket. See the documentation in @network@ for @socket@.+--+-- There is no interruptible variant of this function because it cannot+-- block. (It does not actually perform network activity.)+socket ::+ N.Family -- Family Name (usually AF_INET)+ -> N.SocketType -- Socket Type (usually Stream)+ -> N.ProtocolNumber -- Protocol Number (getProtocolByName to find value)+ -> IO (Either Errno Socket) -- Unconnected Socket+socket !fam !stype !protocol = case stype of+ N.Stream -> finish X.stream+ N.Datagram -> finish X.datagram+ _ -> fail "Network.Unexceptional.socket: Currently only supports stream and datagram types"+ where+ finish !sockTy = mask_ $ do+ X.uninterruptibleSocket (X.Family (N.packFamily fam)) (X.applySocketFlags (X.closeOnExec <> X.nonblocking) sockTy) (X.Protocol protocol) >>= \case+ Left err -> pure (Left err)+ Right (Fd fd) -> do+ s <- mkSocket fd+ pure (Right s)
src/Network/Unexceptional/Bytes.hs view
@@ -6,26 +6,35 @@ module Network.Unexceptional.Bytes ( send+ , sendInterruptible , receive+ , receiveInterruptible ) where +import Control.Applicative ((<|>))+import Control.Concurrent.STM (STM,TVar) import Control.Exception (throwIO)+import Control.Monad ((<=<)) import Control.Monad (when) import Data.Array.Byte (ByteArray) import Data.Bytes.Types (Bytes(Bytes),MutableBytes(MutableBytes))+import Data.Functor (($>))+import Data.Primitive (MutableByteArray) import Foreign.C.Error (Errno) import Foreign.C.Error.Pattern (pattern EWOULDBLOCK,pattern EAGAIN)-import GHC.Conc (threadWaitWrite)+import GHC.Conc (threadWaitWrite,threadWaitWriteSTM)+import GHC.Exts (RealWorld) import Network.Socket (Socket) import System.Posix.Types (Fd(Fd)) -import qualified Network.Unexceptional.Types as Types-import qualified Posix.Socket as X-import qualified Linux.Socket as X+import qualified Control.Concurrent.STM as STM import qualified Data.Bytes.Types-import qualified Network.Socket as S import qualified Data.Primitive as PM+import qualified Linux.Socket as X+import qualified Network.Socket as S import qualified Network.Unexceptional.MutableBytes as MB+import qualified Network.Unexceptional.Types as Types+import qualified Posix.Socket as X -- | Send the entire byte sequence. This call POSIX @send@ in a loop -- until all of the bytes have been sent.@@ -60,6 +69,49 @@ LT -> sendLoop fd arr (off + sentSz) (len - sentSz) GT -> fail "Network.Unexceptional.Bytes.sendAll: send claimed to send too many bytes" +sendInterruptibleLoop :: TVar Bool -> Fd -> ByteArray -> Int -> Int -> IO (Either Errno ())+sendInterruptibleLoop !interrupt !fd !arr !off !len =+ X.uninterruptibleSendByteArray fd arr off (fromIntegral len) (X.noSignal <> X.dontWait) >>= \case+ Left e -> if e == EAGAIN || e == EWOULDBLOCK+ then waitUntilWriteable interrupt fd >>= \case+ Ready -> sendInterruptibleLoop interrupt fd arr off len+ Interrupted -> pure (Left EAGAIN)+ else pure (Left e)+ Right sentSzC ->+ let sentSz = fromIntegral sentSzC :: Int+ in case compare sentSz len of+ EQ -> pure (Right ())+ LT -> sendInterruptibleLoop interrupt fd arr (off + sentSz) (len - sentSz)+ GT -> fail "Network.Unexceptional.Bytes.sendAll: send claimed to send too many bytes"++-- | Variant of 'send' that fails with @EAGAIN@ if the interrupt ever becomes true.+sendInterruptible ::+ TVar Bool+ -> Socket+ -> Bytes+ -> IO (Either Errno ())+sendInterruptible !interrupt s Bytes{array,offset,length=len} = case len of+ 0 -> pure (Right ())+ _ -> STM.readTVarIO interrupt >>= \case+ True -> pure (Left EAGAIN)+ False -> S.withFdSocket s $ \fd ->+ -- We attempt the first send without testing if the socket is in+ -- ready for writes. This is because it is uncommon for the transmit+ -- buffer to already be full.+ sendInterruptibleLoop interrupt (Fd fd) array offset len++data Outcome = Ready | Interrupted++checkFinished :: TVar Bool -> STM ()+checkFinished = STM.check <=< STM.readTVar++waitUntilWriteable :: TVar Bool -> Fd -> IO Outcome+waitUntilWriteable !interrupt !fd = do+ (isReadyAction,deregister) <- threadWaitWriteSTM fd+ outcome <- STM.atomically $ (isReadyAction $> Ready) <|> (checkFinished interrupt $> Interrupted)+ deregister+ pure outcome+ -- | If this returns zero bytes, it means that the peer has -- performed an orderly shutdown. receive :: @@ -69,12 +121,26 @@ receive s n = if n > 0 then do dst <- PM.newByteArray n- MB.receive s (MutableBytes dst 0 n) >>= \case- Left e -> pure (Left e)- Right m -> do- when (m < n) (PM.shrinkMutableByteArray dst m)- dst' <- PM.unsafeFreezeByteArray dst - pure (Right (Bytes dst' 0 m))+ MB.receive s (MutableBytes dst 0 n) >>= handleRececeptionResult dst n else throwIO Types.NonpositiveReceptionSize +-- | If this returns zero bytes, it means that the peer has+-- performed an orderly shutdown.+receiveInterruptible :: + TVar Bool -- ^ Interrupt+ -> Socket+ -> Int -- ^ Maximum number of bytes to receive+ -> IO (Either Errno Bytes)+receiveInterruptible !interrupt s n = if n > 0+ then do+ dst <- PM.newByteArray n+ MB.receiveInterruptible interrupt s (MutableBytes dst 0 n) >>= handleRececeptionResult dst n+ else throwIO Types.NonpositiveReceptionSize +handleRececeptionResult :: MutableByteArray RealWorld -> Int -> Either Errno Int -> IO (Either Errno Bytes)+handleRececeptionResult !dst !n x = case x of+ Left e -> pure (Left e)+ Right m -> do+ when (m < n) (PM.shrinkMutableByteArray dst m)+ dst' <- PM.unsafeFreezeByteArray dst + pure (Right (Bytes dst' 0 m))
src/Network/Unexceptional/Chunks.hs view
@@ -6,11 +6,13 @@ module Network.Unexceptional.Chunks ( send+ , sendInterruptible ) where import Foreign.C.Error (Errno) import Network.Socket (Socket) import Data.Bytes.Chunks (Chunks)+import Control.Concurrent.STM (TVar) import qualified Data.Bytes.Chunks as Chunks import qualified Network.Unexceptional.Bytes as NB@@ -22,3 +24,10 @@ -> Chunks -> IO (Either Errno ()) send !s cs = NB.send s (Chunks.concat cs)++sendInterruptible ::+ TVar Bool+ -> Socket+ -> Chunks+ -> IO (Either Errno ())+sendInterruptible !interrupt !s cs = NB.sendInterruptible interrupt s (Chunks.concat cs)
src/Network/Unexceptional/MutableBytes.hs view
@@ -6,22 +6,28 @@ module Network.Unexceptional.MutableBytes ( receive+ , receiveInterruptible ) where +import Control.Applicative ((<|>))+import Control.Concurrent.STM (STM,TVar) import Control.Exception (throwIO)+import Control.Monad ((<=<)) import Data.Bytes.Types (MutableBytes(MutableBytes))+import Data.Functor (($>)) import Data.Primitive (MutableByteArray) import Foreign.C.Error (Errno) import Foreign.C.Error.Pattern (pattern EWOULDBLOCK,pattern EAGAIN)-import GHC.Conc (threadWaitRead)+import GHC.Conc (threadWaitRead,threadWaitReadSTM) import GHC.Exts (RealWorld) import Network.Socket (Socket) import System.Posix.Types (Fd(Fd)) -import qualified Network.Unexceptional.Types as Types+import qualified Control.Concurrent.STM as STM import qualified Data.Bytes.Types import qualified Linux.Socket as X import qualified Network.Socket as S+import qualified Network.Unexceptional.Types as Types import qualified Posix.Socket as X -- | Receive bytes from a socket. Receives at most N bytes, where N@@ -39,6 +45,19 @@ receiveLoop (Fd fd) array offset len else throwIO Types.NonpositiveReceptionSize +receiveInterruptible ::+ TVar Bool -- ^ Interrupt+ -> Socket+ -> MutableBytes RealWorld -- ^ Slice of a buffer+ -> IO (Either Errno Int)+receiveInterruptible !interrupt s MutableBytes{array,offset,length=len} =+ if len > 0+ then S.withFdSocket s $ \fd ->+ -- We attempt the first receive without testing if the socket is+ -- ready for reads.+ receiveInterruptibleLoop interrupt (Fd fd) array offset len+ else throwIO Types.NonpositiveReceptionSize+ -- Does not wait for file descriptor to be ready. Only performs -- a single successful recv syscall receiveLoop :: Fd -> MutableByteArray RealWorld -> Int -> Int -> IO (Either Errno Int)@@ -54,3 +73,32 @@ in case compare recvSz len of GT -> throwIO Types.ReceivedTooManyBytes _ -> pure (Right recvSz)++-- Does not wait for file descriptor to be ready. Only performs+-- a single successful recv syscall+receiveInterruptibleLoop :: TVar Bool -> Fd -> MutableByteArray RealWorld -> Int -> Int -> IO (Either Errno Int)+receiveInterruptibleLoop !interrupt !fd !arr !off !len =+ X.uninterruptibleReceiveMutableByteArray fd arr off (fromIntegral len) X.dontWait >>= \case+ Left e -> if e == EAGAIN || e == EWOULDBLOCK+ then waitUntilReadable interrupt fd >>= \case+ Ready -> receiveInterruptibleLoop interrupt fd arr off len+ Interrupted -> pure (Left EAGAIN)+ else pure (Left e)+ Right recvSzC ->+ let recvSz = fromIntegral recvSzC :: Int+ in case compare recvSz len of+ GT -> throwIO Types.ReceivedTooManyBytes+ _ -> pure (Right recvSz)++checkFinished :: TVar Bool -> STM ()+checkFinished = STM.check <=< STM.readTVar++data Outcome = Ready | Interrupted++waitUntilReadable :: TVar Bool -> Fd -> IO Outcome+waitUntilReadable !interrupt !fd = do+ (isReadyAction,deregister) <- threadWaitReadSTM fd+ outcome <- STM.atomically $ (isReadyAction $> Ready) <|> (checkFinished interrupt $> Interrupted)+ deregister+ pure outcome+