diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,18 @@ # Revision history for stream-sockets -## Unreleased +## 0.3.1.0 -* introduce `receiveMutableByteArraySlice` which allow receiving in an arbitrary buffer slice. +* Add functions for sending strict bytestrings and lazy bytestrings + over a stream socket (`sendByteArray` and `sendLazyByteArray`). +* Add `receiveByteString` to `Socket.Stream.IPv4`. This is a function + for receive strict bytestring. +* Add `sendAddr`. This is used in the implementations of both + `sendByteArray` and `sendLazyByteArray`. + +## 0.3.0.0 + +* Totally redo almost everything. The API is now considered fairly + stable. ## 0.1.0.0 -- 2019-01-18 diff --git a/bench/Macro.hs b/bench/Macro.hs --- a/bench/Macro.hs +++ b/bench/Macro.hs @@ -137,7 +137,7 @@ act !sock !buf@(MutablePrimArray buf#) !remotes !counter !done = forever $ do n <- scramble buf let remote = PM.indexPrimArray remotes (mod n participants) - unhandled $ DIU.sendMutableByteArray sock (Endpoint {port = remote, address = IPv4.loopback}) + unhandled $ DIU.sendMutableByteArraySlice sock (Endpoint {port = remote, address = IPv4.loopback}) (PM.MutableByteArray buf#) 0 (payloadSize * PM.sizeOf (undefined :: Int)) recvSz <- unhandled $ DIU.receiveMutableByteArraySlice_ sock (PM.MutableByteArray buf#) 0 (payloadSize * PM.sizeOf (undefined :: Int)) diff --git a/sockets.cabal b/sockets.cabal --- a/sockets.cabal +++ b/sockets.cabal @@ -1,17 +1,46 @@ cabal-version: 2.2 name: sockets -version: 0.3.0.0 +version: 0.3.1.0 synopsis: High-level network sockets description: This library provides a high-level abstraction for network sockets. It uses Haskell2010 (along with GADTs) without typeclasses to ensure that consumers of the API can only call appropriate functions on a socket. + . Exceptions are tracked in the types of functions and returned to the caller with `Either`. The caller is free to handle these gracefully or to throw - them. This library only throws exceptions when it detects that it has misused - the operating system's sockets API (open an issue for this) or when the - caller asks for a negatively-sized slice of a buffer (such exceptions are - unrecoverable and indicate a mistake in the code consuming this API). + them. This library has another class of exceptions described as _unrecoverable_. + This library only throws exceptions in three situations: + . + * The library detects that it has misused the operating system's + sockets API. This includes getting a `sockaddr` with an unexpected + socket family. It also includes getting an error code that should not + be possible. For example, the abstractions provided for both datagram + sockets and stream sockets mean that `send` system calls in either + context should never return the error code `ENOTCONN`. Consequently, + this error is treated as unrecoverable. + . + * The caller asks for a negatively-sized slice of a buffer + (such exceptions indicate a mistake in the code consuming this API). + . + * A system call fails with `ENOBUFS` or `ENOMEM`. These indicate that + the operating system is out of memory. If this happens, the + Out Of Memory (OOM) manager is likely killing processes to + reclaim memory, so the process that received this message may + be killed soon. Making things even worse is that the GHC runtime + requests pages of memory from the operating system at times that + are effectively unpredictable to Haskell developers. (Most + memory-managed languages have this behavior). Any attempt + to recover from `ENOBUFS` or `ENOMEM` might cause the runtime to + allocate memory from the operating system. According to the + documentation for the + exception, an allocation failure at this point in time (likely given + the recent `ENOBUFS`/`ENOMEM`) would result in immidiate + termination of the program. So, although it is technically possible + to recover from `ENOBUFS`/`ENOMEM`, the OOM killer and the + GHC runtime make it impossible to do so reliably. Consequently, + these error codes are treated as fatal. + homepage: https://github.com/andrewthad/sockets bug-reports: https://github.com/andrewthad/sockets/issues license: BSD-3-Clause @@ -52,6 +81,8 @@ Socket build-depends: , base >= 4.11.1.0 && < 5 + , bytestring >= 0.10 && < 0.11 + , error-codes >= 0.1 && < 0.2 , ip >= 1.4.1 , posix-api >= 0.2.1 , primitive >= 0.6.4 @@ -81,6 +112,7 @@ , ip >= 1.4.1 , primitive >= 0.6.4 , async + , bytestring ghc-options: -Wall -O2 -threaded default-language: Haskell2010 diff --git a/src/Socket/Datagram/IPv4/Spoof.hs b/src/Socket/Datagram/IPv4/Spoof.hs --- a/src/Socket/Datagram/IPv4/Spoof.hs +++ b/src/Socket/Datagram/IPv4/Spoof.hs @@ -49,9 +49,10 @@ import Text.Printf (printf) import qualified Data.Primitive as PM +import qualified Foreign.C.Error.Describe as D +import qualified GHC.Exts as E import qualified Linux.Socket as L import qualified Posix.Socket as S -import qualified GHC.Exts as E import qualified Socket as SCK -- TODO: Something I am not sure about is whether or not it is necessary @@ -277,7 +278,7 @@ [describeErrorCode e] describeErrorCode :: Errno -> String -describeErrorCode (Errno e) = "error code " ++ show e +describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")" handleSendException :: String -> Errno -> IO (Either (SendException i) a) {-# INLINE handleSendException #-} diff --git a/src/Socket/Datagram/IPv4/Undestined.hs b/src/Socket/Datagram/IPv4/Undestined.hs --- a/src/Socket/Datagram/IPv4/Undestined.hs +++ b/src/Socket/Datagram/IPv4/Undestined.hs @@ -49,6 +49,7 @@ import qualified Socket as SCK import qualified Control.Monad.Primitive as PM import qualified Data.Primitive as PM +import qualified Foreign.C.Error.Describe as D import qualified Linux.Socket as L import qualified Posix.Socket as S @@ -363,7 +364,7 @@ functionWithSocket = "withSocket" describeErrorCode :: Errno -> String -describeErrorCode (Errno e) = "error code " ++ show e +describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")" handleSendException :: String -> Errno -> IO (Either (SendException i) a) {-# INLINE handleSendException #-} diff --git a/src/Socket/Stream.hs b/src/Socket/Stream.hs --- a/src/Socket/Stream.hs +++ b/src/Socket/Stream.hs @@ -60,7 +60,7 @@ ConnectFileDescriptorLimit :: ConnectException i -- | The network is unreachable. (@ENETUNREACH@) ConnectNetworkUnreachable :: ConnectException i - -- | All port numbers numbers in the ephemeral port range are currently in + -- | All port numbers in the ephemeral port range are currently in -- use. (@EADDRNOTAVAIL@) ConnectEphemeralPortsExhausted :: ConnectException i -- | No one is listening on the remote address. (@ECONNREFUSED@) diff --git a/src/Socket/Stream/IPv4.hs b/src/Socket/Stream/IPv4.hs --- a/src/Socket/Stream/IPv4.hs +++ b/src/Socket/Stream/IPv4.hs @@ -16,6 +16,7 @@ , withListener , withAccepted , withConnection + , interruptibleWithConnection , forkAccepted , forkAcceptedUnmasked , interruptibleForkAcceptedUnmasked @@ -24,6 +25,9 @@ , sendByteArraySlice , sendMutableByteArray , sendMutableByteArraySlice + , sendAddr + , sendByteString + , sendLazyByteString , interruptibleSendByteArray , interruptibleSendByteArraySlice , interruptibleSendMutableByteArraySlice @@ -31,6 +35,7 @@ , receiveBoundedByteArray , receiveBoundedMutableByteArraySlice , receiveMutableByteArray + , receiveByteString , interruptibleReceiveByteArray , interruptibleReceiveBoundedMutableByteArraySlice -- * Exceptions @@ -47,6 +52,7 @@ , unlisten , unlisten_ , connect + , interruptibleConnect , disconnect , disconnect_ , accept @@ -58,19 +64,19 @@ import Control.Concurrent (threadWaitReadSTM,threadWaitWriteSTM) import Control.Concurrent (forkIO, forkIOWithUnmask) import Control.Exception (mask, mask_, onException, throwIO) -import Control.Monad.STM (STM,atomically,retry) +import Control.Monad.STM (atomically,retry) import Control.Concurrent.STM (TVar,modifyTVar',readTVar) -import Data.Bifunctor (bimap,first) import Data.Bool (bool) +import Data.ByteString (ByteString) import Data.Functor (($>)) -import Data.Primitive (ByteArray, MutableByteArray(..)) +import Data.Primitive (Addr(..), ByteArray, MutableByteArray(..)) import Data.Word (Word16) import Foreign.C.Error (Errno(..), eAGAIN, eINPROGRESS, eWOULDBLOCK, ePIPE, eNOTCONN) import Foreign.C.Error (eADDRINUSE,eCONNRESET) import Foreign.C.Error (eNFILE,eMFILE,eACCES,ePERM,eCONNABORTED) import Foreign.C.Error (eTIMEDOUT,eADDRNOTAVAIL,eNETUNREACH,eCONNREFUSED) import Foreign.C.Types (CInt, CSize) -import GHC.Exts (Int(I#), RealWorld, shrinkMutableByteArray#) +import GHC.Exts (Ptr(Ptr),Int(I#), RealWorld, shrinkMutableByteArray#, byteArrayContents#, unsafeCoerce#) import Net.Types (IPv4(..)) import Socket (Interruptibility(..)) import Socket (SocketUnrecoverableException(..)) @@ -82,10 +88,15 @@ import System.Posix.Types(Fd) import qualified Control.Monad.Primitive as PM +import qualified Data.ByteString.Internal as BI +import qualified Data.ByteString.Unsafe as BU +import qualified Data.ByteString.Lazy.Internal as LBS import qualified Data.Primitive as PM +import qualified Foreign.C.Error.Describe as D import qualified Linux.Socket as L import qualified Posix.Socket as S import qualified Socket as SCK +import qualified GHC.ForeignPtr as FP -- | A socket that listens for incomming connections. newtype Listener = Listener Fd @@ -252,41 +263,6 @@ SCK.functionWithAccepted [SCK.cgetsockname,SCK.socketAddressSize] --- This function factors out the common elements of withAccepted, forkAccepted, --- and forkAcceptedUnmasked. Unfortunately, I can barely understand it. The --- higher-rank callback is particularly impenetrable. Sorry. -internalAccepted :: - ((forall x. IO x -> IO x) -> ((IO a -> IO d) -> IO (Either CloseException (),d)) -> IO (Either (AcceptException 'Uninterruptible) c)) - -> Listener - -> (Connection -> Endpoint -> IO a) - -> IO (Either (AcceptException 'Uninterruptible) c) -internalAccepted wrap (Listener !lst) f = do - threadWaitRead lst - mask $ \restore -> do - S.uninterruptibleAccept lst S.sizeofSocketAddressInternet >>= \case - Left err -> handleAcceptException "withAccepted" err - Right (sockAddrRequiredSz,sockAddr,acpt) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet - then case S.decodeSocketAddressInternet sockAddr of - Just sockAddrInet -> do - let acceptedEndpoint = socketAddressInternetToEndpoint sockAddrInet - debug ("internalAccepted: successfully accepted connection from " ++ show acceptedEndpoint) - wrap restore $ \restore' -> do - a <- onException (restore' (f (Connection acpt) acceptedEndpoint)) (S.uninterruptibleClose acpt) - e <- gracefulCloseA acpt - pure (e,a) - Nothing -> do - _ <- S.uninterruptibleClose acpt - throwIO $ SocketUnrecoverableException - moduleSocketStreamIPv4 - SCK.functionWithAccepted - [SCK.cgetsockname,SCK.nonInternetSocketFamily] - else do - _ <- S.uninterruptibleClose acpt - throwIO $ SocketUnrecoverableException - moduleSocketStreamIPv4 - SCK.functionWithAccepted - [SCK.cgetsockname,SCK.socketAddressSize] - gracefulCloseA :: Fd -> IO (Either CloseException ()) gracefulCloseA fd = S.uninterruptibleShutdown fd S.write >>= \case -- On Linux (not sure about others), calling shutdown @@ -396,13 +372,13 @@ -> (Connection -> Endpoint -> IO a) -- ^ Callback to consume connection. Must not return the connection. -> IO (Either (AcceptException 'Uninterruptible) ThreadId) -forkAccepted lst consumeException cb = internalAccepted - ( \restore action -> do - tid <- forkIO $ do - (e,x) <- action restore - restore (consumeException e x) - pure (Right tid) - ) lst cb +forkAccepted lstn consumeException cb = + mask $ \restore -> accept lstn >>= \case + Left e -> pure (Left e) + Right (conn, endpoint) -> fmap Right $ forkIO $ do + a <- onException (restore (cb conn endpoint)) (disconnect_ conn) + e <- disconnect conn + restore (consumeException e a) -- | Accept a connection on the listener and run the supplied callback in -- a new thread. The masking state is set to @Unmasked@ when running the @@ -521,11 +497,51 @@ -- ^ Remote endpoint -> IO (Either (ConnectException 'Uninterruptible) Connection) connect !remote = do - debug ("connect: opening connection " ++ show remote) + beforeEstablishment remote >>= \case + Left err -> pure (Left err) + Right (fd,sockAddr) -> S.uninterruptibleConnect fd sockAddr >>= \case + Left err2 -> if err2 == eINPROGRESS + then do + threadWaitWrite fd + afterEstablishment fd + else do + S.uninterruptibleErrorlessClose fd + handleConnectException SCK.functionWithConnection err2 + Right _ -> afterEstablishment fd + +-- | Variant of 'connect' that is interruptible using STM-style interrupts. +interruptibleConnect :: + TVar Bool + -- ^ Interrupted. If this becomes 'True', give up and return + -- @'Left' 'AcceptInterrupted'@. + -> Endpoint + -- ^ Remote endpoint + -> IO (Either (ConnectException 'Interruptible) Connection) +interruptibleConnect !abandon !remote = do + beforeEstablishment remote >>= \case + Left err -> pure (Left err) + Right (fd,sockAddr) -> S.uninterruptibleConnect fd sockAddr >>= \case + Left err2 -> if err2 == eINPROGRESS + then do + interruptibleWaitWrite abandon fd >>= \case + True -> afterEstablishment fd + False -> pure (Left ConnectInterrupted) + else do + S.uninterruptibleErrorlessClose fd + handleConnectException SCK.functionWithConnection err2 + Right _ -> afterEstablishment fd + +-- Internal function called by both connect and interruptibleConnect +-- before the connection is established. Creates the socket and prepares +-- the sockaddr. +beforeEstablishment :: Endpoint -> IO (Either (ConnectException i) (Fd,S.SocketAddress)) +{-# INLINE beforeEstablishment #-} +beforeEstablishment !remote = do + debug ("beforeEstablishment: opening connection " ++ show remote) e1 <- S.uninterruptibleSocket S.internet (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream) S.defaultProtocol - debug ("connect: opened connection " ++ show remote) + debug ("beforeEstablishment: opened connection " ++ show remote) case e1 of Left err -> handleSocketConnectException SCK.functionWithConnection err Right fd -> do @@ -533,42 +549,36 @@ $ S.encodeSocketAddressInternet $ endpointToSocketAddressInternet $ remote - merr <- S.uninterruptibleConnect fd sockAddr >>= \case - Left err2 -> if err2 == eINPROGRESS - then do - threadWaitWrite fd - pure Nothing - else pure (Just err2) - Right _ -> pure Nothing - case merr of - Just err -> do - S.uninterruptibleErrorlessClose fd - handleConnectException SCK.functionWithConnection err - Nothing -> do - e <- S.uninterruptibleGetSocketOption fd - S.levelSocket S.optionError (intToCInt (PM.sizeOf (undefined :: CInt))) - case e of - Left err -> do - S.uninterruptibleErrorlessClose fd - throwIO $ SocketUnrecoverableException - moduleSocketStreamIPv4 - functionWithListener - [SCK.cgetsockopt,describeEndpoint remote,describeErrorCode err] - Right (sz,S.OptionValue val) -> if sz == intToCInt (PM.sizeOf (undefined :: CInt)) - then - let err = PM.indexByteArray val 0 :: CInt in - if err == 0 - then pure (Right (Connection fd)) - else do - S.uninterruptibleErrorlessClose fd - handleConnectException SCK.functionWithConnection (Errno err) - else do - S.uninterruptibleErrorlessClose fd - throwIO $ SocketUnrecoverableException - moduleSocketStreamIPv4 - functionWithListener - [SCK.cgetsockopt,describeEndpoint remote,connectErrorOptionValueSize] + pure (Right (fd,sockAddr)) +-- Internal function called by both connect and interruptibleConnect +-- after the connection is established. +afterEstablishment :: Fd -> IO (Either (ConnectException i) Connection) +afterEstablishment fd = do + e <- S.uninterruptibleGetSocketOption fd + S.levelSocket S.optionError (intToCInt (PM.sizeOf (undefined :: CInt))) + case e of + Left err -> do + S.uninterruptibleErrorlessClose fd + throwIO $ SocketUnrecoverableException + moduleSocketStreamIPv4 + functionWithListener + [SCK.cgetsockopt,describeErrorCode err] + Right (sz,S.OptionValue val) -> if sz == intToCInt (PM.sizeOf (undefined :: CInt)) + then + let err = PM.indexByteArray val 0 :: CInt in + if err == 0 + then pure (Right (Connection fd)) + else do + S.uninterruptibleErrorlessClose fd + handleConnectException SCK.functionWithConnection (Errno err) + else do + S.uninterruptibleErrorlessClose fd + throwIO $ SocketUnrecoverableException + moduleSocketStreamIPv4 + functionWithListener + [SCK.cgetsockopt,connectErrorOptionValueSize] + -- | Close a connection gracefully, reporting a 'CloseException' when -- the connection has to be terminated by sending a TCP reset. This -- uses a combination of @shutdown@, @recv@, @close@ to detect when @@ -605,6 +615,26 @@ m <- disconnect conn b <- g m a pure (Right b) + +interruptibleWithConnection :: + TVar Bool + -- ^ Interrupted. If this becomes 'True', give up and return + -- @'Left' 'AcceptInterrupted'@. + -> Endpoint + -- ^ Remote endpoint + -> (Either CloseException () -> a -> IO b) + -- ^ Callback to handle an ungraceful close. + -> (Connection -> IO a) + -- ^ Callback to consume connection. Must not return the connection. + -> IO (Either (ConnectException 'Interruptible) b) +interruptibleWithConnection !abandon !remote g f = mask $ \restore -> do + interruptibleConnect abandon remote >>= \case + Left err -> pure (Left err) + Right conn -> do + a <- onException (restore (f conn)) (disconnect_ conn) + m <- disconnect conn + b <- g m a + pure (Right b) sendByteArray :: Connection -- ^ Connection @@ -623,6 +653,45 @@ interruptibleSendByteArray abandon conn arr = interruptibleSendByteArraySlice abandon conn arr 0 (PM.sizeofByteArray arr) +-- | Send a 'ByteString' over a connection. +sendByteString :: + Connection -- ^ Connection + -> ByteString -- ^ Payload + -> IO (Either (SendException 'Uninterruptible) ()) +sendByteString !conn !payload = BU.unsafeUseAsCStringLen payload + (\(Ptr addr,len) -> sendAddr conn (Addr addr) len) + +-- | Send a lazy 'LBS.ByteString' over a connection. +sendLazyByteString :: + Connection -- ^ Connection + -> LBS.ByteString -- ^ Payload + -> IO (Either (SendException 'Uninterruptible) ()) +sendLazyByteString !conn !chunks0 = go chunks0 where + go LBS.Empty = pure (Right ()) + go (LBS.Chunk chunk chunks) = sendByteString conn chunk >>= \case + Left e -> pure (Left e) + Right _ -> go chunks + +sendAddr :: + Connection -- ^ Connection + -> Addr -- ^ Payload start address + -> Int -- ^ Payload length + -> IO (Either (SendException 'Uninterruptible) ()) +sendAddr !conn !payload0 !len0 = go payload0 len0 + where + go !payload !len = if len > 0 + then internalSendAddr conn payload len >>= \case + Left e -> pure (Left e) + Right sz' -> do + let sz = csizeToInt sz' + go (PM.plusAddr payload sz) (len - sz) + else if len == 0 + then pure (Right ()) + else throwIO $ SocketUnrecoverableException + moduleSocketStreamIPv4 + functionSendByteArray + [SCK.negativeSliceLength] + sendByteArraySlice :: Connection -- ^ Connection -> ByteArray -- ^ Payload (will be sliced) @@ -791,6 +860,16 @@ conn payload off len -- Precondition: the length must be greater than zero. +internalSendAddr :: + Connection -- ^ Connection + -> Addr -- ^ Buffer start address + -> Int -- ^ Length of slice into buffer + -> IO (Either (SendException 'Uninterruptible) CSize) +internalSendAddr !conn !payload !len = veryInternalSendAddr + (\fd -> threadWaitWrite fd *> pure (Right ())) + conn payload len + +-- Precondition: the length must be greater than zero. internalInterruptibleSend :: TVar Bool -- ^ Aliveness -> Connection -- ^ Connection @@ -839,6 +918,38 @@ else handleSendException functionSendByteArray err1 Right sz -> pure (Right sz) +-- Precondition: the length must be greater than zero. +veryInternalSendAddr :: + (Fd -> IO (Either (SendException i) ())) + -> Connection -- ^ Connection + -> Addr -- ^ Buffer start address + -> Int -- ^ Length of payload + -> IO (Either (SendException i) CSize) +{-# INLINE veryInternalSendAddr #-} +veryInternalSendAddr wait (Connection !s) !payload !len = do + debug ("veryInternalSendAddr: about to send chunk on stream socket, length " ++ show len) + e1 <- S.uninterruptibleSend s payload + (intToCSize len) + (S.noSignal) + debug "veryInternalSendAddr: just sent chunk on stream socket" + case e1 of + Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN + then do + debug "veryInternalSendAddr: waiting to for write ready on stream socket" + wait s >>= \case + Left e -> pure (Left e) + Right _ -> do + e2 <- S.uninterruptibleSend s payload + (intToCSize len) + (S.noSignal) + case e2 of + Left err2 -> do + debug "veryInternalSendAddr: encountered error after sending chunk on stream socket" + handleSendException functionSendByteArray err2 + Right sz -> pure (Right sz) + else handleSendException functionSendByteArray err1 + Right sz -> pure (Right sz) + -- The maximum number of bytes to receive must be greater than zero. -- The operating system guarantees us that the returned actual number -- of bytes is less than or equal to the requested number of bytes. @@ -927,6 +1038,17 @@ functionReceiveByteArray [SCK.negativeSliceLength] +-- | Receive exactly the given number of bytes. +receiveByteString :: + Connection -- ^ Connection + -> Int -- ^ Number of bytes to receive + -> IO (Either (ReceiveException 'Uninterruptible) ByteString) +receiveByteString !conn !total = do + marr@(MutableByteArray marr#) <- PM.newPinnedByteArray total + receiveMutableByteArray conn marr >>= \case + Left err -> pure (Left err) + Right _ -> pure (Right (BI.PS (FP.ForeignPtr (byteArrayContents# (unsafeCoerce# marr#)) (FP.PlainPtr marr#)) 0 total)) + -- | Receive a number of bytes exactly equal to the size of the mutable -- byte array. If the remote application shuts down its end of the -- connection before sending the required number of bytes, this returns @@ -1064,7 +1186,7 @@ functionReceiveMutableByteArraySlice = "receiveMutableByteArraySlice" describeErrorCode :: Errno -> String -describeErrorCode (Errno e) = "error code " ++ show e +describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")" handleReceiveException :: String -> Errno -> IO (Either (ReceiveException i) a) handleReceiveException func e diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -11,6 +11,7 @@ import Control.Exception (throwIO) import Control.Monad.ST (runST) import Data.Bool (bool) +import Data.ByteString (ByteString) import Data.Primitive (ByteArray) import Data.Word (Word16,Word8) import GHC.Exts (RealWorld) @@ -19,6 +20,7 @@ import Test.Tasty import Test.Tasty.HUnit +import qualified Data.ByteString as B import qualified Data.Primitive as PM import qualified Data.Primitive.MVar as PM import qualified GHC.Exts as E @@ -63,6 +65,7 @@ , testCase "4MB" (testStreamB 4) , testCase "32MB" (testStreamB 32) ] + , testCase "C" testStreamC ] ] ] @@ -204,6 +207,24 @@ let theSize = PM.indexByteArray serializedSize 0 :: Int result <- unhandled $ SI.receiveByteArray conn theSize pure result + +testStreamC :: Assertion +testStreamC = do + (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar + ((),received) <- concurrently (sender m) (receiver m) + received @=? message + where + message = B.pack (enumFromTo 0 (100 :: Word8)) :: ByteString + sender :: PM.MVar RealWorld Word16 -> IO () + sender m = do + dstPort <- PM.takeMVar m + unhandled $ SI.withConnection (DIU.Endpoint IPv4.loopback dstPort) unhandledClose $ \conn -> do + unhandled $ SI.sendByteString conn message + receiver :: PM.MVar RealWorld Word16 -> IO ByteString + receiver m = unhandled $ SI.withListener (SI.Endpoint IPv4.loopback 0) $ \listener port -> do + PM.putMVar m port + unhandled $ SI.withAccepted listener unhandledClose $ \conn _ -> do + unhandled $ SI.receiveByteString conn (B.length message) -- The sender sends a large amount of traffic that may exceed -- the size of the operating system's TCP send buffer. The