sockets 0.3.0.0 → 0.3.1.0
raw patch · 8 files changed
+283/−96 lines, 8 filesdep +error-codesdep ~bytestring
Dependencies added: error-codes
Dependency ranges changed: bytestring
Files
- CHANGELOG.md +12/−2
- bench/Macro.hs +1/−1
- sockets.cabal +37/−5
- src/Socket/Datagram/IPv4/Spoof.hs +3/−2
- src/Socket/Datagram/IPv4/Undestined.hs +2/−1
- src/Socket/Stream.hs +1/−1
- src/Socket/Stream/IPv4.hs +206/−84
- test/Main.hs +21/−0
CHANGELOG.md view
@@ -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
bench/Macro.hs view
@@ -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))
sockets.cabal view
@@ -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 <http://hackage.haskell.org/package/base-4.12.0.0/docs/Control-Exception-Base.html#t:AsyncException HeapOverflow>+ 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
src/Socket/Datagram/IPv4/Spoof.hs view
@@ -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 #-}
src/Socket/Datagram/IPv4/Undestined.hs view
@@ -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 #-}
src/Socket/Stream.hs view
@@ -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@)
src/Socket/Stream/IPv4.hs view
@@ -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
test/Main.hs view
@@ -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