packages feed

sockets 0.4.0.0 → 0.5.0.0

raw patch · 19 files changed

+857/−32 lines, 19 filesdep ~posix-apidep ~stm

Dependency ranges changed: posix-api, stm

Files

CHANGELOG.md view
@@ -1,5 +1,18 @@ # Revision history for stream-sockets +## 0.5.0.0++* Add support for getting back `ECONNREFUSED` when send or receiving+  with datagram sockets.+* Add support for using `Addr` and `ByteString` with datagram sockets.+* Add sendMany for vectored io on a stream socket.+* Correct the implementation of interruptible functions that+  work on datagram sockets.++## 0.4.0.0++* Overhaul the interface. Aggressively use backpack.+ ## 0.3.1.0  * Add functions for sending strict bytestrings and lazy bytestrings
sockets.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: sockets-version: 0.4.0.0+version: 0.5.0.0 synopsis: High-level network sockets description:   This library provides a high-level abstraction for network sockets. It uses@@ -79,6 +79,7 @@ library sockets-internal   build-depends:     , base >= 4.11.1.0 && <5+    , bytestring >=0.10 && <0.11     , ip >= 1.4.1     , posix-api >=0.3 && <0.4     , primitive-offset >= 0.2 && <0.3@@ -94,6 +95,7 @@   exposed-modules:     Socket     Socket.AddrLength+    Socket.Connected.UnmanagedBytes     Socket.Datagram     Socket.Debug     Socket.Error@@ -101,6 +103,7 @@     Socket.IPv4     Socket.Stream     Socket.Bytes+    Socket.Destined.IPv4.UnmanagedBytes     Socket.Destined.IPv4.MutableBytes     Socket.Destined.IPv4.Bytes     Socket.Discard@@ -109,8 +112,11 @@     Socket.MutableBytes     Socket.MutableBytes.Peerless     Socket.MutableBytes.SocketAddressInternet+    Socket.UnmanagedBytes.Peerless+    Socket.UnmanagedBytes.SocketAddressInternet     Socket.Interruptible     Socket.Uninterruptible+    Socket.Interop     Hybrid.Send.MutableBytes.AddrLength   if flag(debug)     hs-source-dirs: src-debug@@ -290,10 +296,14 @@     Socket.Datagram.IPv4.Unconnected     Socket.Datagram.IPv4.Connected     Socket.Stream.IPv4+    Socket.Datagram.Interruptible.Addr     Socket.Datagram.Interruptible.Bytes+    Socket.Datagram.Interruptible.ByteString     Socket.Datagram.Interruptible.MutableBytes     Socket.Datagram.Slab+    Socket.Datagram.Uninterruptible.Addr     Socket.Datagram.Uninterruptible.Bytes+    Socket.Datagram.Uninterruptible.ByteString     Socket.Datagram.Uninterruptible.MutableBytes     Socket.Stream.Interruptible.Addr     Socket.Stream.Interruptible.ByteString@@ -336,6 +346,18 @@        Socket.Interrupt as Socket.Uninterruptible,        Socket.Buffer as Socket.MutableBytes),     sockets-datagram-send+      (Datagram.Send.Indefinite as Socket.Datagram.Uninterruptible.Addr.Send.IPv4)+      requires+      (Datagram.Send as Socket.Destined.IPv4.UnmanagedBytes,+       Socket.Interrupt as Socket.Uninterruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-send+      (Datagram.Send.Indefinite as Socket.Datagram.Interruptible.Addr.Send.IPv4)+      requires+      (Datagram.Send as Socket.Destined.IPv4.UnmanagedBytes,+       Socket.Interrupt as Socket.Interruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-send       (Datagram.Send.Indefinite as Socket.Datagram.Uninterruptible.Bytes.Send.Connected)       requires       (Datagram.Send as Socket.Connected.Bytes,@@ -365,6 +387,18 @@       (Datagram.Send as Socket.Connected.MutableBytes,        Socket.Interrupt as Socket.Interruptible,        Socket.Buffer as Socket.MutableBytes),+    sockets-datagram-send+      (Datagram.Send.Indefinite as Socket.Datagram.Interruptible.Addr.Send.Connected)+      requires+      (Datagram.Send as Socket.Connected.UnmanagedBytes,+       Socket.Interrupt as Socket.Interruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-send+      (Datagram.Send.Indefinite as Socket.Datagram.Uninterruptible.Addr.Send.Connected)+      requires+      (Datagram.Send as Socket.Connected.UnmanagedBytes,+       Socket.Interrupt as Socket.Uninterruptible,+       Socket.Buffer as Socket.AddrLength),     sockets-datagram-receive       (Datagram.Receive.Indefinite as Socket.Datagram.Uninterruptible.MutableBytes.Receive.IPv4)       requires@@ -372,6 +406,18 @@        Socket.Interrupt as Socket.Uninterruptible,        Socket.Buffer as Socket.MutableBytes),     sockets-datagram-receive+      (Datagram.Receive.Indefinite as Socket.Datagram.Uninterruptible.Addr.Receive.IPv4)+      requires+      (Datagram.Receive as Socket.UnmanagedBytes.SocketAddressInternet,+       Socket.Interrupt as Socket.Uninterruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-receive+      (Datagram.Receive.Indefinite as Socket.Datagram.Interruptible.Addr.Receive.IPv4)+      requires+      (Datagram.Receive as Socket.UnmanagedBytes.SocketAddressInternet,+       Socket.Interrupt as Socket.Interruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-receive       (Datagram.Receive.Indefinite as Socket.Datagram.Interruptible.MutableBytes.Receive.IPv4)       requires       (Datagram.Receive as Socket.MutableBytes.SocketAddressInternet,@@ -389,6 +435,18 @@       (Datagram.Receive as Socket.MutableBytes.Peerless,        Socket.Interrupt as Socket.Interruptible,        Socket.Buffer as Socket.MutableBytes),+    sockets-datagram-receive+      (Datagram.Receive.Indefinite as Socket.Datagram.Interruptible.Addr.Receive.Connected)+      requires+      (Datagram.Receive as Socket.UnmanagedBytes.Peerless,+       Socket.Interrupt as Socket.Interruptible,+       Socket.Buffer as Socket.AddrLength),+    sockets-datagram-receive+      (Datagram.Receive.Indefinite as Socket.Datagram.Uninterruptible.Addr.Receive.Connected)+      requires+      (Datagram.Receive as Socket.UnmanagedBytes.Peerless,+       Socket.Interrupt as Socket.Uninterruptible,+       Socket.Buffer as Socket.AddrLength),     sockets-stream-send       (Stream.Send.Indefinite as Socket.Stream.Interruptible.Bytes.Send)       requires@@ -495,6 +553,7 @@     , byteslice >= 0.1.1     , primitive-unlifted     , primitive-addr+    , stm   if flag(checked)     build-depends: primitive-checked >= 0.7 && <0.8   else
src-datagram-receive/Datagram/Receive/Indefinite.hs view
@@ -9,7 +9,7 @@   ) where  import Control.Concurrent.STM (TVar)-import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK)+import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK, eCONNREFUSED) import Foreign.C.Types (CSize) import Socket.Error (die) import Socket.EventManager (Token)@@ -55,12 +55,13 @@   -- we return an exception.   e <- Receive.receiveFromOnce fd buf L.truncate addrBuf   case e of-    Left err -> if err == eWOULDBLOCK || err == eAGAIN-      then pure (Right Nothing)-      else die $ concat-        [ "Socket.Datagram.receive: " -        , describeErrorCode err-        ]+    Left err ->+      if | err == eWOULDBLOCK || err == eAGAIN -> pure (Right Nothing)+         | err == eCONNREFUSED -> pure (Left ReceiveConnectionRefused)+         | otherwise -> die $ concat+             [ "Socket.Datagram.receive: " +             , describeErrorCode err+             ]     Right recvSz -> do       let !recvSzInt = csizeToInt recvSz       if recvSzInt <= Buffer.length buf@@ -81,7 +82,9 @@     Nothing -> do       EM.unready token0 tv       token1 <- wait intr tv-      receiveLoop intr tv token1 fd buf addrBuf+      case tokenToDatagramReceiveException token1 of+        Left err -> pure (Left err)+        Right _ -> receiveLoop intr tv token1 fd buf addrBuf     Just !r -> pure (Right r)  csizeToInt :: CSize -> Int
src-datagram-send/Datagram/Send/Indefinite.hs view
@@ -9,6 +9,7 @@ import Control.Concurrent.STM (TVar) import Datagram.Send (Peer) import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK, eACCES)+import Foreign.C.Error (eCONNREFUSED) import Foreign.C.Types (CSize) import Socket.Error (die) import Socket.EventManager (Token)@@ -26,7 +27,12 @@ -- POSIX @send@. This is used for datagram sockets. We cannot use a -- Socket newtype here since destined and undestined sockets -- use different newtypes.-send :: Interrupt -> Peer -> Fd -> Buffer -> IO (Either (SendException Intr) ())+send ::+     Interrupt+  -> Peer+  -> Fd+  -> Buffer+  -> IO (Either (SendException Intr) ()) send !intr !dst !sock !buf = do   let !mngr = EM.manager   tv <- EM.writer mngr sock@@ -48,6 +54,7 @@                Left err -> pure (Left err)                Right _ -> sendLoop intr dst sock tv new buf          | e == eACCES -> pure (Left SendBroadcasted)+         | e == eCONNREFUSED -> pure (Left SendConnectionRefused)          | otherwise -> die ("Socket.Datagram.send: " ++ describeErrorCode e)     Right sz -> if csizeToInt sz == Buffer.length buf       then pure $! Right ()
+ src-internal/Socket/Connected/UnmanagedBytes.hs view
@@ -0,0 +1,34 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language NamedFieldPuns #-}++module Socket.Connected.UnmanagedBytes+  ( Buffer+  , Peer+  , advance+  , length+  , send+  ) where++import Prelude hiding (length)++import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Socket.AddrLength (Buffer,length,advance)+import Foreign.C.Error (Errno)+import Foreign.C.Types (CInt,CSize)+import System.Posix.Types (Fd)+import qualified Posix.Socket as S++type Peer = ()++send :: () -> Fd -> Buffer -> IO (Either Errno CSize)+send !_ !sock (UnmanagedBytes arr len) =+  -- No need for MSG_NOSIGNAL since this is a datagram+  -- socket, not a stream socket.+  S.uninterruptibleSend sock arr+    (intToCSize len)+    mempty++intToCSize :: Int -> CSize+intToCSize = fromIntegral+
src-internal/Socket/Datagram.hs view
@@ -27,6 +27,14 @@   SendTruncated :: !Int -> SendException i   -- | Attempted to send to a broadcast address.   SendBroadcasted :: SendException i+  -- | A previous attempt to send to the peer failed, and+  --   the failure is asynchronously reported via ICMP on a later+  --   attempt to send to the peer. It is likely that the+  --   peer is not listening on the requested port or that the+  --   datagram could not be routed to the peer. (This is a+  --   confusing way to notify the user of a send failure, but+  --   it is how Linux does it.)+  SendConnectionRefused :: SendException i   -- | STM-style interrupt (much safer than C-style interrupt)   SendInterrupted :: SendException 'Interruptible @@ -34,12 +42,26 @@ deriving stock instance Eq (SendException i) deriving anyclass instance (Typeable i) => Exception (SendException i) +-- | Exceptions that can happen while attempting to receive+--   a datagram.  data ReceiveException :: Interruptibility -> Type where   -- | The datagram did not fit in the buffer. The field is the   --   original size of the datagram that was truncated. If   --   this happens, the process probably needs to start using   --   a larger receive buffer.   ReceiveTruncated :: !Int -> ReceiveException i+  -- | A previous attempt to send to the peer failed, and+  --   the failure is asynchronously reported via ICMP on a later+  --   attempt to receive from the peer. It is likely that the+  --   peer is not listening on the requested port or that the+  --   datagram could not be routed to the peer. Even if the user+  --   has not sent anything, then this exception can result from+  --   a previous socket bound to the same port. In this case,+  --   it can be safely ignored. However, if the user has sent+  --   something over the socket, this exception means that it+  --   likely did not arrive. (This is a confusing way to notify+  --   the user of a send failure, but it is how Linux does it.)+  ReceiveConnectionRefused :: ReceiveException i   -- | STM-style interrupt (much safer than C-style interrupt)   ReceiveInterrupted :: ReceiveException 'Interruptible
+ src-internal/Socket/Destined/IPv4/UnmanagedBytes.hs view
@@ -0,0 +1,46 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language NamedFieldPuns #-}++module Socket.Destined.IPv4.UnmanagedBytes+  ( Buffer+  , Peer+  , advance+  , length+  , send+  ) where++import Prelude hiding (length)++import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Foreign.C.Error (Errno)+import Foreign.C.Types (CInt,CSize)+import GHC.Exts (RealWorld)+import Net.Types (IPv4(..))+import Socket.IPv4 (Peer(..))+import Socket.AddrLength (advance,length)+import System.Posix.Types (Fd)+import qualified Posix.Socket as S++type Buffer = UnmanagedBytes++send :: Peer -> Fd -> UnmanagedBytes -> IO (Either Errno CSize)+send !dst !sock (UnmanagedBytes arr len) =+  S.uninterruptibleSendToInternet sock arr+    (intToCSize len)+    mempty+    (endpointToSocketAddressInternet dst)++endpointToSocketAddressInternet :: Peer -> S.SocketAddressInternet+endpointToSocketAddressInternet (Peer {address, port}) =+  S.SocketAddressInternet+    { S.port = S.hostToNetworkShort port+    , S.address = S.hostToNetworkLong (getIPv4 address)+    }++intToCInt :: Int -> CInt+intToCInt = fromIntegral++intToCSize :: Int -> CSize+intToCSize = fromIntegral+
+ src-internal/Socket/Interop.hs view
@@ -0,0 +1,16 @@+{-# language MagicHash #-}++module Socket.Interop+  ( fromPinned+  ) where++import Data.ByteString.Internal (ByteString(PS))+import Data.Primitive (MutableByteArray(..))+import GHC.Exts (RealWorld,byteArrayContents#,unsafeCoerce#)+import GHC.ForeignPtr (ForeignPtr(ForeignPtr),ForeignPtrContents(PlainPtr))++-- Precondition: the argument buffer is pinned+fromPinned :: MutableByteArray RealWorld -> Int -> Int -> ByteString+{-# inline fromPinned #-}+fromPinned (MutableByteArray marr#) off len =+  PS (ForeignPtr (byteArrayContents# (unsafeCoerce# marr#)) (PlainPtr marr#)) off len
+ src-internal/Socket/UnmanagedBytes/Peerless.hs view
@@ -0,0 +1,44 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}++module Socket.UnmanagedBytes.Peerless+  ( AddressBuffer+  , AddressBufferOffset+  , writeAddress+  , offsetAddress+  , receiveFromOnce+  ) where++import Prelude hiding (length)++import Data.Bytes.Types (UnmanagedBytes(..))+import Foreign.C.Error (Errno)+import Foreign.C.Types (CSize)+import Posix.Socket (MessageFlags,Message(Receive))+import System.Posix.Types (Fd)+import qualified Posix.Socket as S++type Address = ()+type AddressBuffer = ()+type AddressBufferOffset = ()++writeAddress :: AddressBuffer -> Int -> Address -> IO ()+writeAddress _ _ _ = pure ()++offsetAddress :: AddressBuffer -> Int -> AddressBufferOffset+offsetAddress _ _ = ()++receiveFromOnce ::+     Fd+  -> UnmanagedBytes+  -> MessageFlags 'Receive+  -> ()+  -> IO (Either Errno CSize)+{-# inline receiveFromOnce #-}+receiveFromOnce fd (UnmanagedBytes arr len) flags !_ =+  S.uninterruptibleReceiveFrom_+    fd arr (intToCSize len) flags++intToCSize :: Int -> CSize+{-# inline intToCSize #-}+intToCSize = fromIntegral
+ src-internal/Socket/UnmanagedBytes/SocketAddressInternet.hs view
@@ -0,0 +1,50 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}++module Socket.UnmanagedBytes.SocketAddressInternet+  ( AddressBuffer+  , AddressBufferOffset+  , writeAddress+  , offsetAddress+  , receiveFromOnce+  ) where++import Prelude hiding (length)++import Data.Bytes.Types (UnmanagedBytes(..))+import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset(..))+import Foreign.C.Error (Errno)+import Foreign.C.Types (CSize)+import GHC.Exts (RealWorld)+import Posix.Socket (MessageFlags,Message(Receive))+import Posix.Socket (SocketAddressInternet)+import System.Posix.Types (Fd)++import qualified Data.Primitive as PM+import qualified Posix.Socket as S++type AddressBuffer = PM.MutablePrimArray RealWorld SocketAddressInternet+type AddressBufferOffset = MutablePrimArrayOffset RealWorld SocketAddressInternet++writeAddress :: AddressBuffer -> Int -> SocketAddressInternet -> IO ()+{-# inline writeAddress #-}+writeAddress = PM.writePrimArray++offsetAddress :: AddressBuffer -> Int -> AddressBufferOffset+{-# inline offsetAddress #-}+offsetAddress = MutablePrimArrayOffset++receiveFromOnce ::+     Fd+  -> UnmanagedBytes+  -> MessageFlags 'Receive+  -> MutablePrimArrayOffset RealWorld SocketAddressInternet+  -> IO (Either Errno CSize)+{-# inline receiveFromOnce #-}+receiveFromOnce !sock (UnmanagedBytes arr len) !flags !addr =+  S.uninterruptibleReceiveFromInternet sock arr (intToCSize len) flags addr+   +intToCSize :: Int -> CSize+{-# inline intToCSize #-}+intToCSize = fromIntegral+
src/Socket/Datagram/IPv4/Connected.hs view
@@ -13,12 +13,16 @@ module Socket.Datagram.IPv4.Connected   ( -- * Types     Socket(..)+  , Family(..)+  , Connectedness(..)   , Peer(..)   , Message(..)     -- * Establish   , withSocket     -- * Exceptions   , SocketException(..)+  , SD.ReceiveException(..)+  , SD.SendException(..)     -- * Examples     -- $examples   ) where@@ -29,7 +33,7 @@ import Foreign.C.Error (eNFILE,eMFILE,eADDRINUSE) import GHC.IO (IO(..)) import Net.Types (IPv4(..))-import Socket (Connectedness(Connected))+import Socket (Connectedness(..),Family(..)) import Socket.Datagram (Socket(..)) import Socket.Datagram (SocketException(..)) import Socket.IPv4 (Peer(..),Message(..))@@ -40,6 +44,7 @@ import qualified Posix.Socket as S import qualified Socket as SCK import qualified Socket.EventManager as EM+import qualified Socket.Datagram as SD  withSocket ::      Peer
src/Socket/Datagram/IPv4/Unconnected.hs view
@@ -12,6 +12,8 @@ module Socket.Datagram.IPv4.Unconnected   ( -- * Types     Socket(..)+  , Family(..)+  , Connectedness(..)   , Peer(..)   , Message(..)     -- * Establish@@ -26,7 +28,7 @@ import Foreign.C.Error (eNFILE,eMFILE,eADDRINUSE) import GHC.IO (IO(..)) import Net.Types (IPv4(..))-import Socket (Connectedness(Unconnected))+import Socket (Connectedness(..),Family(..)) import Socket.Datagram (Socket(..)) import Socket.Datagram (SocketException(..)) import Socket.Debug (debug)
+ src/Socket/Datagram/Interruptible/Addr.hs view
@@ -0,0 +1,88 @@+{-# language BangPatterns #-}+{-# language LambdaCase #-}+{-# language GADTSyntax #-}+{-# language KindSignatures #-}+{-# language DataKinds #-}+{-# language MagicHash #-}++module Socket.Datagram.Interruptible.Addr+  ( -- * Send+    send+  , sendToIPv4+    -- * Receive+  , receive+  , receiveFromIPv4+    -- * Slabs+    -- ** Types+  , PeerlessSlab(..)+  , IPv4Slab(..)+    -- ** Functions+  , newPeerlessSlab+  , newIPv4Slab+  ) where++import Control.Concurrent.STM (TVar)+import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Data.Primitive.Addr (Addr)+import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset)+import GHC.Exts (RealWorld)+import Posix.Socket (SocketAddressInternet)+import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Interruptible))+import Socket.Datagram (Socket(..),SendException,ReceiveException)+import Socket.IPv4 (Peer,newIPv4Slab,IPv4Slab(..))+import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)++import qualified Socket.Datagram.Interruptible.Addr.Receive.Connected as CR+import qualified Socket.Datagram.Interruptible.Addr.Send.Connected as CS+import qualified Socket.Datagram.Interruptible.Addr.Send.IPv4 as V4S+import qualified Socket.Datagram.Interruptible.Addr.Receive.IPv4 as V4R++-- | Send a datagram using a socket with a pre-designated peer. This+-- refers to a datagram socket for which POSIX @connect@ has locked+-- down communication to an individual peer.+send ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Connected a -- ^ Socket with designated peer+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Number of bytes to send+  -> IO (Either (SendException 'Interruptible) ())+send !intr (Socket !sock) !buf !len =+  CS.send intr () sock (UnmanagedBytes buf len)++-- | Receive a datagram, discarding the peer address. This can be used with+-- datagram sockets of any family. It is usable with both connected and+-- unconnected datagram sockets.+receive ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket c a -- ^ Socket+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Interruptible) Int)+receive !intr (Socket !sock) !buf !len =+  CR.receive intr sock (UnmanagedBytes buf len) ()++sendToIPv4 ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Peer -- ^ Destination+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Number of bytes to send+  -> IO (Either (SendException 'Interruptible) ())+sendToIPv4 !intr (Socket !sock) !dst !buf !len =+  V4S.send intr dst sock (UnmanagedBytes buf len)++receiveFromIPv4 ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Maximum number of bytes to receive+  -> MutablePrimArrayOffset RealWorld SocketAddressInternet+     -- ^ Buffer for returned peer address+  -> IO (Either (ReceiveException 'Interruptible) Int)+receiveFromIPv4 !intr (Socket !sock) !buf !len !addr =+  V4R.receive intr sock (UnmanagedBytes buf len) addr+
+ src/Socket/Datagram/Interruptible/ByteString.hs view
@@ -0,0 +1,99 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language GADTSyntax #-}+{-# language KindSignatures #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language ScopedTypeVariables #-}++module Socket.Datagram.Interruptible.ByteString+  ( -- * Send+    send+  , sendToIPv4+    -- * Receive+  , receive+  , receiveFromIPv4+    -- * Slabs+    -- ** Types+  , PeerlessSlab(..)+  , IPv4Slab(..)+    -- ** Functions+  , newPeerlessSlab+  , newIPv4Slab+  ) where++import Control.Concurrent.STM (TVar)+import Data.ByteString (ByteString)+import Data.ByteString.Unsafe (unsafeUseAsCStringLen)+import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Data.Bytes.Types (MutableBytes(MutableBytes))+import Data.Primitive.Addr (Addr(..))+import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset)+import GHC.Exts (Ptr(Ptr),RealWorld)+import Posix.Socket (SocketAddressInternet)+import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Interruptible))+import Socket.Datagram (Socket(..),SendException,ReceiveException)+import Socket.IPv4 (Peer,newIPv4Slab,IPv4Slab(..))+import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)+import Socket.Interop (fromPinned)++import qualified Data.Primitive as PM+import qualified Socket.Datagram.Interruptible.MutableBytes.Receive.Connected as CR+import qualified Socket.Datagram.Interruptible.Addr.Send.Connected as CS+import qualified Socket.Datagram.Interruptible.Addr.Send.IPv4 as V4S+import qualified Socket.Datagram.Interruptible.MutableBytes.Receive.IPv4 as V4R++-- | Send a datagram using a socket with a pre-designated peer. This+-- refers to a datagram socket for which POSIX @connect@ has locked+-- down communication to an individual peer.+send ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Connected a -- ^ Socket with designated peer+  -> ByteString -- ^ Slice of a buffer+  -> IO (Either (SendException 'Interruptible) ())+send !intr (Socket !sock) !bs = unsafeUseAsCStringLen bs+  (\(Ptr addr#,len) ->+    CS.send intr () sock (UnmanagedBytes (Addr addr#) len)+  )++-- | Receive a datagram, discarding the peer address. This can be used with+-- datagram sockets of any family. It is usable with both connected and+-- unconnected datagram sockets.+receive ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket c a -- ^ Socket+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Interruptible) ByteString)+receive !intr (Socket !sock) !len = do+  !marr <- PM.newPinnedByteArray len+  CR.receive intr sock (MutableBytes marr 0 len) () >>= \case+    Right sz -> pure $! Right $! fromPinned marr 0 sz+    Left err -> pure (Left err)++sendToIPv4 ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Peer -- ^ Destination+  -> ByteString -- ^ Slice of a buffer+  -> IO (Either (SendException 'Interruptible) ())+sendToIPv4 !intr (Socket !sock) !dst !bs = unsafeUseAsCStringLen bs+  (\(Ptr addr#,len) ->+    V4S.send intr dst sock (UnmanagedBytes (Addr addr#) len)+  )++receiveFromIPv4 ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return @'Left' 'SendInterrupted'@.+  -> Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Int -- ^ Maximum number of bytes to receive+  -> MutablePrimArrayOffset RealWorld SocketAddressInternet+     -- ^ Buffer for returned peer address+  -> IO (Either (ReceiveException 'Interruptible) ByteString)+receiveFromIPv4 !intr (Socket !sock) !len !addr = do+  !marr <- PM.newPinnedByteArray len+  V4R.receive intr sock (MutableBytes marr 0 len) addr >>= \case+    Left err -> pure (Left err)+    Right sz -> pure $! Right $! fromPinned marr 0 sz
+ src/Socket/Datagram/Uninterruptible/Addr.hs view
@@ -0,0 +1,79 @@+{-# language BangPatterns #-}+{-# language LambdaCase #-}+{-# language GADTSyntax #-}+{-# language KindSignatures #-}+{-# language DataKinds #-}+{-# language MagicHash #-}++module Socket.Datagram.Uninterruptible.Addr+  ( -- * Send+    send+  , sendToIPv4+    -- * Receive+  , receive+  , receiveFromIPv4+    -- * Slabs+    -- ** Types+  , PeerlessSlab(..)+  , IPv4Slab(..)+    -- ** Functions+  , newPeerlessSlab+  , newIPv4Slab+  ) where++import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Data.Primitive.Addr (Addr)+import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset)+import GHC.Exts (RealWorld,proxy#)+import Posix.Socket (SocketAddressInternet)+import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Uninterruptible))+import Socket.Datagram (Socket(..),SendException,ReceiveException)+import Socket.IPv4 (Peer,newIPv4Slab,IPv4Slab(..))+import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)++import qualified Socket.Datagram.Uninterruptible.Addr.Receive.Connected as CR+import qualified Socket.Datagram.Uninterruptible.Addr.Send.Connected as CS+import qualified Socket.Datagram.Uninterruptible.Addr.Send.IPv4 as V4S+import qualified Socket.Datagram.Uninterruptible.Addr.Receive.IPv4 as V4R++-- | Send a datagram using a socket with a pre-designated peer. This+-- refers to a datagram socket for which POSIX @connect@ has locked+-- down communication to an individual peer.+send ::+     Socket 'Connected a -- ^ Socket with designated peer+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Number of bytes to send+  -> IO (Either (SendException 'Uninterruptible) ())+send (Socket !sock) !buf !len =+  CS.send proxy# () sock (UnmanagedBytes buf len)++-- | Receive a datagram, discarding the peer address. This can be used with+-- datagram sockets of any family. It is usable with both connected and+-- unconnected datagram sockets.+receive ::+     Socket c a -- ^ Socket+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Uninterruptible) Int)+receive (Socket !sock) !buf !len =+  CR.receive proxy# sock (UnmanagedBytes buf len) ()++sendToIPv4 ::+     Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Peer -- ^ Destination+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Number of bytes to send+  -> IO (Either (SendException 'Uninterruptible) ())+sendToIPv4 (Socket !sock) !dst !buf !len =+  V4S.send proxy# dst sock (UnmanagedBytes buf len)++receiveFromIPv4 ::+     Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Addr -- ^ Start of buffer+  -> Int -- ^ Maximum number of bytes to receive+  -> MutablePrimArrayOffset RealWorld SocketAddressInternet+     -- ^ Buffer for returned peer address+  -> IO (Either (ReceiveException 'Uninterruptible) Int)+receiveFromIPv4 (Socket !sock) !buf !len !addr =+  V4R.receive proxy# sock (UnmanagedBytes buf len) addr+
+ src/Socket/Datagram/Uninterruptible/ByteString.hs view
@@ -0,0 +1,90 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language GADTSyntax #-}+{-# language KindSignatures #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language ScopedTypeVariables #-}++module Socket.Datagram.Uninterruptible.ByteString+  ( -- * Send+    send+  , sendToIPv4+    -- * Receive+  , receive+  , receiveFromIPv4+    -- * Slabs+    -- ** Types+  , PeerlessSlab(..)+  , IPv4Slab(..)+    -- ** Functions+  , newPeerlessSlab+  , newIPv4Slab+  ) where++import Data.ByteString (ByteString)+import Data.ByteString.Unsafe (unsafeUseAsCStringLen)+import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes))+import Data.Bytes.Types (MutableBytes(MutableBytes))+import Data.Primitive.Addr (Addr(..))+import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset)+import GHC.Exts (Ptr(Ptr),RealWorld,proxy#)+import Posix.Socket (SocketAddressInternet)+import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Uninterruptible))+import Socket.Datagram (Socket(..),SendException,ReceiveException)+import Socket.IPv4 (Peer,newIPv4Slab,IPv4Slab(..))+import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)+import Socket.Interop (fromPinned)++import qualified Data.Primitive as PM+import qualified Socket.Datagram.Uninterruptible.MutableBytes.Receive.Connected as CR+import qualified Socket.Datagram.Uninterruptible.Addr.Send.Connected as CS+import qualified Socket.Datagram.Uninterruptible.Addr.Send.IPv4 as V4S+import qualified Socket.Datagram.Uninterruptible.MutableBytes.Receive.IPv4 as V4R++-- | Send a datagram using a socket with a pre-designated peer. This+-- refers to a datagram socket for which POSIX @connect@ has locked+-- down communication to an individual peer.+send ::+     Socket 'Connected a -- ^ Socket with designated peer+  -> ByteString -- ^ Slice of a buffer+  -> IO (Either (SendException 'Uninterruptible) ())+send (Socket !sock) !bs = unsafeUseAsCStringLen bs+  (\(Ptr addr#,len) ->+    CS.send proxy# () sock (UnmanagedBytes (Addr addr#) len)+  )++-- | Receive a datagram, discarding the peer address. This can be used with+-- datagram sockets of any family. It is usable with both connected and+-- unconnected datagram sockets.+receive ::+     Socket c a -- ^ Socket+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Uninterruptible) ByteString)+receive (Socket !sock) !len = do+  !marr <- PM.newPinnedByteArray len+  CR.receive proxy# sock (MutableBytes marr 0 len) () >>= \case+    Right sz -> pure $! Right $! fromPinned marr 0 sz+    Left err -> pure (Left err)++sendToIPv4 ::+     Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Peer -- ^ Destination+  -> ByteString -- ^ Slice of a buffer+  -> IO (Either (SendException 'Uninterruptible) ())+sendToIPv4 (Socket !sock) !dst !bs = unsafeUseAsCStringLen bs+  (\(Ptr addr#,len) ->+    V4S.send proxy# dst sock (UnmanagedBytes (Addr addr#) len)+  )++receiveFromIPv4 ::+     Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer+  -> Int -- ^ Maximum number of bytes to receive+  -> MutablePrimArrayOffset RealWorld SocketAddressInternet+     -- ^ Buffer for returned peer address+  -> IO (Either (ReceiveException 'Uninterruptible) ByteString)+receiveFromIPv4 (Socket !sock) !len !addr = do+  !marr <- PM.newPinnedByteArray len+  V4R.receive proxy# sock (MutableBytes marr 0 len) addr >>= \case+    Left err -> pure (Left err)+    Right sz -> pure $! Right $! fromPinned marr 0 sz
src/Socket/Stream/Uninterruptible/ByteString.hs view
@@ -1,7 +1,8 @@ {-# language BangPatterns #-}-{-# language LambdaCase #-} {-# language DataKinds #-}+{-# language LambdaCase #-} {-# language MagicHash #-}+{-# language ScopedTypeVariables #-}  module Socket.Stream.Uninterruptible.ByteString   ( send@@ -10,16 +11,15 @@   , receiveBetween   ) where +import Data.ByteString (ByteString) import Data.Bytes.Types (UnmanagedBytes(..)) import Data.ByteString.Unsafe (unsafeUseAsCStringLen)-import Data.ByteString.Internal (ByteString(PS)) import Data.Primitive.Addr (Addr(..))-import Data.Primitive (MutableByteArray(..)) import Data.Bytes.Types (MutableBytes(..))-import GHC.Exts (Ptr(Ptr),RealWorld,byteArrayContents#,unsafeCoerce#,proxy#)-import GHC.ForeignPtr (ForeignPtr(ForeignPtr),ForeignPtrContents(PlainPtr))+import GHC.Exts (Ptr(Ptr),proxy#) import Socket.Stream (Connection,ReceiveException,SendException) import Socket (Interruptibility(Uninterruptible))+import Socket.Interop (fromPinned)  import qualified Data.Primitive as PM import qualified Socket.Stream.Uninterruptible.Addr.Send as Send@@ -47,7 +47,7 @@   !marr <- PM.newPinnedByteArray n   Receive.receiveExactly proxy# conn (MutableBytes marr 0 n) >>= \case     Left err -> pure (Left err)-    Right _ -> pure $! Right $! fromManaged marr 0 n+    Right (_ :: ()) -> pure $! Right $! fromPinned marr 0 n  -- | Receive at most the specified number of bytes. This -- only makes multiple calls to POSIX @recv@ if EAGAIN is returned. It makes at@@ -61,9 +61,7 @@   !marr0 <- PM.newPinnedByteArray n   Receive.receiveOnce proxy# conn (MutableBytes marr0 0 n) >>= \case     Left err -> pure (Left err)-    Right sz -> do-      marr1 <- PM.resizeMutableByteArray marr0 sz-      pure $! Right $! fromManaged marr1 0 sz+    Right sz -> pure $! Right $! fromPinned marr0 0 sz  -- | Receive a number of bytes that is between the inclusive lower and --   upper bounds. If needed, this may call @recv@ repeatedly until the@@ -78,12 +76,4 @@   !marr0 <- PM.newPinnedByteArray maxLen   Receive.receiveBetween proxy# conn (MutableBytes marr0 0 maxLen) minLen >>= \case     Left err -> pure (Left err)-    Right sz -> do-      marr1 <- PM.resizeMutableByteArray marr0 sz-      pure $! Right $! fromManaged marr1 0 sz--fromManaged :: MutableByteArray RealWorld -> Int -> Int -> ByteString-{-# inline fromManaged #-}-fromManaged (MutableByteArray marr#) off len =-  PS (ForeignPtr (byteArrayContents# (unsafeCoerce# marr#)) (PlainPtr marr#)) off len-+    Right sz -> pure $! Right $! fromPinned marr0 0 sz
src/Socket/Stream/Uninterruptible/Bytes.hs view
@@ -2,6 +2,8 @@ {-# language DataKinds #-} {-# language LambdaCase #-} {-# language MagicHash #-}+{-# language MultiWayIf #-}+{-# language TypeApplications #-}  -- | Communicate over a connection using immutable byte arrays. -- Reception functions return 'ByteArray' instead of 'Bytes' since@@ -11,21 +13,38 @@ -- payload. module Socket.Stream.Uninterruptible.Bytes   ( send+  , sendMany   , receiveExactly   , receiveOnce   , receiveBetween   ) where -import Data.Primitive (ByteArray) import Data.Bytes.Types (MutableBytes(..),Bytes(..))+import Data.Primitive (ByteArray) import GHC.Exts (proxy#)-import Socket.Stream (Connection,ReceiveException,SendException) import Socket (Interruptibility(Uninterruptible))+import Socket.Stream (Connection,ReceiveException,SendException)  import qualified Data.Primitive as PM import qualified Socket.Stream.Uninterruptible.Bytes.Send as Send import qualified Socket.Stream.Uninterruptible.MutableBytes.Receive as Receive +-- Used for sendMany+import Control.Monad (when)+import Socket.Stream (SendException(..),Connection(..))+import Data.Primitive.Unlifted.Array (UnliftedArray)+import Socket.EventManager (Token)+import Control.Concurrent.STM (TVar)+import System.Posix.Types (Fd)+import Foreign.C.Types (CSize)+import Socket.Error (die)+import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK, ePIPE, eCONNRESET)+import Socket.Debug (whenDebugging)+import qualified Foreign.C.Error.Describe as D+import qualified Socket.EventManager as EM+import qualified Data.Primitive.Unlifted.Array as PM+import qualified Posix.Socket as S+ -- | Send a slice of a buffer. If needed, this calls POSIX @send@ repeatedly --   until the entire contents of the buffer slice have been sent. send ::@@ -85,4 +104,65 @@       marr1 <- PM.resizeMutableByteArray marr0 sz       !arr <- PM.unsafeFreezeByteArray marr1       pure $! Right $! arr++-- | Send many buffers with vectored I/O. This uses @sendmsg@ to send+--   multiple chunks at a time. It may call @sendmsg@ more than once+--   if there is not enough space in the operating system send buffer+--   to house all the chunks at the same time.+sendMany ::+     Connection -- ^ Connection+  -> UnliftedArray ByteArray -- ^ Byte arrays+  -> IO (Either (SendException 'Uninterruptible) ())+{-# inline sendMany #-}+sendMany (Connection conn) bufs = do+  -- TODO: Generalize sendMany so that an interruptible variant+  -- can be provided.+  let !mngr = EM.manager+  tv <- EM.writer mngr conn+  -- Do not forget to use the correct wait function when this+  -- is generalized.+  token0 <- EM.wait tv+  sendManyLoop conn tv token0 bufs 0 (PM.sizeofUnliftedArray bufs) 0++sendManyLoop ::+     Fd -> TVar Token -> Token+  -> UnliftedArray ByteArray+  -> Int -> Int -> Int+  -> IO (Either (SendException 'Uninterruptible) ())+sendManyLoop !conn !tv !old !bufs !chunkIx !chunkCount !chunkOff = if chunkCount > 0+  then S.uninterruptibleSendByteArrays conn bufs chunkIx chunkCount chunkOff S.noSignal >>= \case+    Left e ->+      if | e == eAGAIN || e == eWOULDBLOCK -> do+             EM.unready old tv+             -- Do not forget to use the correct wait function when this+             -- is generalized.+             new <- EM.wait tv+             sendManyLoop conn tv new bufs chunkIx chunkCount chunkOff+         | e == ePIPE -> pure (Left SendShutdown)+         | e == eCONNRESET -> pure (Left SendReset)+         | otherwise -> die ("Socket.Stream.sendMany: " ++ describeErrorCode e)+    Right sz' -> do+      let sz0 = fromIntegral @CSize @Int sz'+      -- We enumerate the sent chunks, one by one,+      -- to compute the new chunk index.+      -- ix: chunk index+      -- m: offset into the current chunk+      -- n: remaining bytes to account for+      let go !ix !m !n = if n > 0+            then+              let !fullSz = PM.sizeofByteArray (PM.indexUnliftedArray bufs ix)+                  !sz = fullSz - m+               in if sz <= n+                    then go (ix + 1) 0 (n - sz)+                    else (ix, m + n)+            else (ix,0)+          (nextChunkIx,nextChunkOff) = go chunkIx chunkOff sz0+      sendManyLoop conn tv old bufs nextChunkIx (chunkCount - (nextChunkIx - chunkIx)) nextChunkOff+  -- chunkCount must be 0 in this case since it is never negative+  else do+    whenDebugging $ when (chunkCount < 0) $ die "sendManyLoop: negative chunks"+    pure (Right ())++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"
test/Main.hs view
@@ -9,6 +9,7 @@  import Control.Applicative (liftA3) import Control.Concurrent (forkIO)+import Control.Concurrent.STM (newTVarIO) import Control.Concurrent.Async (concurrently) import Control.Monad (replicateM_) import Control.Exception (Exception)@@ -19,6 +20,7 @@ import Data.Bytes.Types (Bytes(..),MutableBytes(..),UnmanagedBytes(..)) import Data.Primitive (ByteArray,MutableByteArray(..)) import Data.Primitive.Addr (Addr(..))+import Data.Primitive.Unlifted.Array (UnliftedArray) import Data.Word (Word16,Word8) import GHC.Exts (RealWorld) import GHC.IO (IO(..))@@ -36,6 +38,7 @@ import qualified Socket.Datagram.IPv4.Unconnected as DIU import qualified Socket.Datagram.IPv4.Connected as DIC import qualified Socket.Datagram.Uninterruptible.Bytes as DUB+import qualified Socket.Datagram.Interruptible.Bytes as DIB import qualified Socket.Stream.IPv4 as SI import qualified Socket.Stream.Uninterruptible.Bytes as UB import qualified Socket.Stream.Uninterruptible.MutableBytes as UMB@@ -56,9 +59,11 @@         , testCase "D" testDatagramUndestinedD         , testCase "E" testDatagramUndestinedE         , testCase "F" testDatagramUndestinedF+        , testCase "G" testDatagramUndestinedG         ]       , testGroup "connected"         [ testCase "A" testDatagramConnectedA+        , testCase "B" testDatagramConnectedB         ]       ]     ]@@ -79,6 +84,8 @@         , testCase "12MB" (testStreamF (4 * 1024))         , testCase "48MB" (testStreamF (16 * 1024))         ]+      , testCase "G" testStreamG+      , testCase "H" testStreamH       ]     ]   ]@@ -121,6 +128,18 @@     PM.putMVar m port     unhandled $ DUB.receiveFromIPv4 sock sz +testDatagramConnectedB :: Assertion+testDatagramConnectedB = do+  intr <- newTVarIO True+  unhandled $ DIC.withSocket+    (DIU.Peer IPv4.loopback 0)+    (DIU.Peer IPv4.loopback 43245)+    (\sock _ -> DIB.receive intr sock 42 >>= \case+      Left DIC.ReceiveInterrupted -> pure ()+      Left e -> throwIO e+      Right _ -> fail "testDatagramUndestinedB: received datagram"+    )+ testDatagramUndestinedA :: Assertion testDatagramUndestinedA = do   (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar@@ -294,6 +313,23 @@     slab <- DUB.newPeerlessSlab 1 (sz - 1)     DUB.receiveMany sock slab +testDatagramUndestinedG :: Assertion+testDatagramUndestinedG = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  ((),received) <- concurrently (sender m) (receiver m)+  received @=? message+  where+  message = E.fromList [0,1,2,3] :: ByteArray+  sz = PM.sizeofByteArray message+  sender :: PM.MVar RealWorld Word16 -> IO ()+  sender m = unhandled $ DIU.withSocket (DUB.Peer IPv4.loopback 0) $ \sock _ -> do+    dstPort <- PM.takeMVar m+    unhandled $ DUB.sendToIPv4 sock (DIU.Peer IPv4.loopback dstPort) (unsliced message)+  receiver :: PM.MVar RealWorld Word16 -> IO ByteArray+  receiver m = unhandled $ DIU.withSocket (DIU.Peer IPv4.loopback 0) $ \sock port -> do+    PM.putMVar m port+    unhandled $ DUB.receive sock sz+ -- This test involves a made up protocol that goes like this: -- The sender always starts by sending the length of the rest -- of the payload as a native-endian encoded machine-sized int.@@ -504,6 +540,68 @@       y <- unhandled $ UMB.receiveBetween conn (MutableBytes marr x 150) 100       unhandled $ UMB.receiveExactly conn (MutableBytes marr (x + y) (256 - (x + y)))       PM.unsafeFreezeByteArray marr++testStreamG :: Assertion+testStreamG = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  ((),received) <- concurrently (sender m) (receiver m)+  received @=? (mconcat [messageA, messageB, messageC, messageD])+  where+  messageA = E.fromList [0..17] :: ByteArray+  messageB = E.fromList [18..92] :: ByteArray+  messageC = E.fromList [93..182] :: ByteArray+  messageD = E.fromList [183..255] :: ByteArray+  sender :: PM.MVar RealWorld Word16 -> IO ()+  sender m = do+    dstPort <- PM.takeMVar m+    unhandled $ SI.withConnection (DIU.Peer IPv4.loopback dstPort) unhandledClose $ \conn -> do+      let msgs = E.fromList [messageA,messageB,messageC,messageD]+      unhandled $ UB.sendMany conn msgs+  receiver :: PM.MVar RealWorld Word16 -> IO ByteArray+  receiver m = unhandled $ SI.withListener (SI.Peer IPv4.loopback 0) $ \listener port -> do+    PM.putMVar m port+    unhandled $ SI.withAccepted listener unhandledClose $ \conn _ -> do+      marr <- PM.newByteArray 256+      x <- unhandled $ UMB.receiveBetween conn (MutableBytes marr 0 60) 20+      y <- unhandled $ UMB.receiveBetween conn (MutableBytes marr x 150) 100+      unhandled $ UMB.receiveExactly conn (MutableBytes marr (x + y) (256 - (x + y)))+      PM.unsafeFreezeByteArray marr++-- The sender sends a large amount of traffic that may exceed+-- the size of the operating system's TCP send buffer, currently+-- 32MB. This uses sendmsg.+testStreamH :: Assertion+testStreamH = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  ((),()) <- concurrently (sender m) (receiver m)+  pure ()+  where+  recvChunkSize = 32 * 1024+  message = E.fromList (replicate (1024 * 256) magicByte) :: ByteArray+  messages = E.fromList (replicate 128 message) :: UnliftedArray ByteArray+  sender :: PM.MVar RealWorld Word16 -> IO ()+  sender m = do+    dstPort <- PM.takeMVar m+    unhandled $ SI.withConnection (DIU.Peer IPv4.loopback dstPort) unhandledClose $ \conn -> do+      unhandled $ UB.sendMany conn messages+  receiver :: PM.MVar RealWorld Word16 -> IO ()+  receiver m = unhandled $ SI.withListener (SI.Peer IPv4.loopback 0) $ \listener port -> do+    PM.putMVar m port+    unhandled $ SI.withAccepted listener unhandledClose $ \conn _ -> do+      buffer <- PM.newByteArray recvChunkSize+      let receiveLoop !remaining+            | remaining > 0 = do+                let recvSize = min remaining recvChunkSize+                PM.setByteArray buffer 0 recvChunkSize (0 :: Word8)+                bytesReceived <- unhandled+                  (UMB.receiveOnce conn (MutableBytes buffer 0 recvSize))+                verifyClientSendBytes buffer bytesReceived >>= \case+                  True -> receiveLoop (remaining - bytesReceived)+                  False -> throwIO MagicByteMismatch+            | remaining == 0 = pure ()+            | otherwise = throwIO NegativeByteCount+      receiveLoop (32 * 1024 * 1024)+      pure ()  touchMutableByteArray :: MutableByteArray RealWorld -> IO () touchMutableByteArray (MutableByteArray x) = touchMutableByteArray# x