packages feed

sockets 0.5.0.0 → 0.6.1.0

raw patch · 40 files changed

+2600/−303 lines, 40 filesdep ~byteslicedep ~bytestringdep ~posix-api

Dependency ranges changed: byteslice, bytestring, posix-api, primitive

Files

CHANGELOG.md view
@@ -1,5 +1,17 @@ # Revision history for stream-sockets +## 0.6.1.0++* Correct implementation of withListenerReuse+* Add interruptibleForkAcceptedUnmasked for UDS. Add `SO_REUSEADDR` inet functions.++## 0.6.0.0++* Give slab types a phantom type argument that indicates whether+  or not they operate on pinned byte arrays. This makes it possible+  to provide a `bytestring` interface to `recvmmsg`.+* Upgrade interruptible datagram interface to work with new slabs.+ ## 0.5.0.0  * Add support for getting back `ECONNREFUSED` when send or receiving
sockets.cabal view
@@ -1,6 +1,6 @@-cabal-version: 2.2+cabal-version: 3.0 name: sockets-version: 0.5.0.0+version: 0.6.1.0 synopsis: High-level network sockets description:   This library provides a high-level abstraction for network sockets. It uses@@ -53,7 +53,7 @@  flag mmsg   manual: True-  description: Use sendmmsg and recvmmsg +  description: Use sendmmsg and recvmmsg   default: False  flag verbose-errors@@ -63,12 +63,14 @@  flag debug   manual: True-  description: Print debug output +  description: Print debug output   default: False  flag checked   manual: True-  description: Add bounds-checking to primitive array operations+  description:+    Add bounds-checking to primitive array operations and to+    certain operations that expect pinned byte arrays.   default: False  flag example@@ -79,15 +81,15 @@ library sockets-internal   build-depends:     , base >= 4.11.1.0 && <5-    , bytestring >=0.10 && <0.11+    , bytestring >=0.10 && <0.12     , ip >= 1.4.1-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , primitive-offset >= 0.2 && <0.3     , primitive-unlifted >= 0.1 && <0.2     , primitive-addr >= 0.1 && <0.2     , stm >= 2.4     , text >= 1.2-    , byteslice >=0.1.1 && <0.2+    , byteslice >=0.1.1 && <0.3   if flag(checked)     build-depends: primitive-checked >= 0.7 && <0.8   else@@ -101,7 +103,9 @@     Socket.Error     Socket.EventManager     Socket.IPv4+    Socket.SequencedPacket     Socket.Stream+    Socket.Slab     Socket.Bytes     Socket.Destined.IPv4.UnmanagedBytes     Socket.Destined.IPv4.MutableBytes@@ -152,7 +156,7 @@   build-depends:     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , stm >= 2.4     , sockets-internal     , sockets-buffer@@ -170,7 +174,7 @@     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2     , stm >= 2.4-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , primitive-offset >= 0.2 && <0.3     , sockets-internal     , sockets-buffer@@ -191,7 +195,7 @@     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2     , stm >= 2.4-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , primitive-unlifted >= 0.1 && <0.2     , byteslice >= 0.1.1     , sockets-datagram-receive@@ -215,13 +219,14 @@   build-depends:     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , stm >= 2.4     , sockets-internal     , sockets-buffer     , sockets-interrupt   exposed-modules:     Stream.Send.Indefinite+    SequencedPacket.Send.Indefinite   signatures:     Stream.Send   hs-source-dirs: src-stream-send@@ -232,13 +237,14 @@   build-depends:     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , stm >= 2.4     , sockets-internal     , sockets-buffer     , sockets-interrupt   exposed-modules:     Stream.Receive.Indefinite+    SequencedPacket.Receive.Indefinite   signatures:     Stream.Receive   hs-source-dirs: src-stream-receive@@ -249,7 +255,7 @@   build-depends:     , base >= 4.11.1.0 && <5     , error-codes >= 0.1.0.1 && < 0.2-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , stm >= 2.4     , sockets-internal     , sockets-interrupt@@ -285,17 +291,24 @@   reexported-modules:     , Stream.Send.Indefinite     , Stream.Receive.Indefinite+    , SequencedPacket.Receive.Indefinite   mixins:       sockets-stream-send (Stream.Send.Indefinite)-    , sockets-stream-receive (Stream.Receive.Indefinite)+    , sockets-stream-receive (Stream.Receive.Indefinite,SequencedPacket.Receive.Indefinite)   default-language: Haskell2010+  ghc-options: -O2 -Wall  library   exposed-modules:     Socket.Address     Socket.Datagram.IPv4.Unconnected     Socket.Datagram.IPv4.Connected+    Socket.Datagram.Unix.Unconnected+    Socket.Datagram.Unix.Connected+    Socket.SequencedPacket.Unix+    Socket.SequencedPacket.Uninterruptible.Bytes     Socket.Stream.IPv4+    Socket.Stream.Unix     Socket.Datagram.Interruptible.Addr     Socket.Datagram.Interruptible.Bytes     Socket.Datagram.Interruptible.ByteString@@ -316,8 +329,11 @@     Socket.Stream.Uninterruptible.MutableBytes     Socket.Stream.Uninterruptible.Hybrid   other-modules:+    Socket.Pair+    Socket.Datagram.Common     Socket.Datagram.Interruptible.MutableBytes.Many     Socket.Datagram.Uninterruptible.MutableBytes.Many+    Socket.Systemd   mixins:     sockets-datagram-receive-many       (Datagram.Receive.Many.Indefinite as Socket.Datagram.Uninterruptible.MutableBytes.Receive.Many.Unit)@@ -454,7 +470,9 @@        Socket.Interrupt as Socket.Interruptible,        Socket.Buffer as Socket.Bytes),     sockets-stream-send-      (Stream.Send.Indefinite as Socket.Stream.Uninterruptible.Bytes.Send)+      (Stream.Send.Indefinite as Socket.Stream.Uninterruptible.Bytes.Send,+       SequencedPacket.Send.Indefinite as Socket.SequencedPacket.Uninterruptible.Bytes.Send+      )       requires       (Stream.Send as Socket.Bytes,        Socket.Interrupt as Socket.Uninterruptible,@@ -469,7 +487,8 @@        Socket.Interrupt as Socket.Interruptible),     sockets-stream-bidirectional       (Stream.Send.Indefinite as Socket.Stream.Uninterruptible.MutableBytes.Send,-       Stream.Receive.Indefinite as Socket.Stream.Uninterruptible.MutableBytes.Receive)+       Stream.Receive.Indefinite as Socket.Stream.Uninterruptible.MutableBytes.Receive,+       SequencedPacket.Receive.Indefinite as Socket.SequencedPacket.Uninterruptible.MutableBytes.Receive)       requires       (Stream.Send as Socket.MutableBytes,        Stream.Receive as Socket.MutableBytes,@@ -509,11 +528,11 @@        Socket.Interrupt as Socket.Uninterruptible),   build-depends:     , base >= 4.11.1.0 && <5-    , byteslice >=0.1.1 && <0.2-    , bytestring >=0.10 && <0.11+    , byteslice >=0.1.1 && <0.3+    , bytestring >=0.10 && <0.12     , error-codes >=0.1.0.1 && <0.2     , ip >=1.4.1-    , posix-api >=0.3 && <0.4+    , posix-api >=0.4 && <0.5     , primitive-addr >= 0.1 && <0.2     , primitive-offset >= 0.2 && <0.3     , primitive-unlifted >= 0.1 && <0.2@@ -599,7 +618,7 @@   if flag(example)     build-depends:       , base >= 4.11.1.0 && < 5-      , byteslice >= 0.1.1 && < 0.2+      , byteslice >= 0.1.1 && < 0.3       , bytestring >= 0.10.8.2       , fast-logger >= 2.4.13       , ip >= 1.4.1
src-datagram-send/Datagram/Send/Indefinite.hs view
@@ -4,12 +4,13 @@  module Datagram.Send.Indefinite   ( send+  , attemptSend   ) where  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.Error (Errno(..), eAGAIN, eWOULDBLOCK, eACCES, eMSGSIZE)+import Foreign.C.Error (eCONNREFUSED,eNOTCONN) import Foreign.C.Types (CSize) import Socket.Error (die) import Socket.EventManager (Token)@@ -41,6 +42,24 @@     Left err -> pure (Left err)     Right _ -> sendLoop intr dst sock tv token0 buf +-- Never blocks.+attemptSend ::+     Peer+  -> Fd+  -> Buffer+  -> IO (Either (SendException Intr) Bool)+attemptSend !dst !sock !buf = Send.send dst sock buf >>= \case+  Left e ->+    if | e == eAGAIN || e == eWOULDBLOCK -> pure (Right False)+       | e == eACCES -> pure (Left SendBroadcasted)+       | e == eCONNREFUSED -> pure (Left SendConnectionRefused)+       | e == eNOTCONN -> pure (Left SendConnectionRefused)+       | e == eMSGSIZE -> pure (Left SendMessageSize)+       | otherwise -> die ("Socket.Datagram.send: " ++ describeErrorCode e)+  Right sz -> if csizeToInt sz == Buffer.length buf+    then pure $! Right True+    else pure $! Left $! SendTruncated $! csizeToInt sz+ sendLoop ::      Interrupt -> Peer -> Fd -> TVar Token -> Token   -> Buffer -> IO (Either (SendException Intr) ())@@ -55,6 +74,8 @@                Right _ -> sendLoop intr dst sock tv new buf          | e == eACCES -> pure (Left SendBroadcasted)          | e == eCONNREFUSED -> pure (Left SendConnectionRefused)+         | e == eNOTCONN -> pure (Left SendConnectionRefused)+         | e == eMSGSIZE -> pure (Left SendMessageSize)          | otherwise -> die ("Socket.Datagram.send: " ++ describeErrorCode e)     Right sz -> if csizeToInt sz == Buffer.length buf       then pure $! Right ()
src-internal/Socket.hs view
@@ -5,11 +5,15 @@ {-# language DuplicateRecordFields #-} {-# language GADTs #-} {-# language KindSignatures #-}+{-# language StandaloneDeriving #-}  module Socket   ( SocketUnrecoverableException(..)+  , SocketException(..)+  , BindException(..)   , Direction(..)   , Connectedness(..)+  , Pinnedness(..)   , Family(..)   , Version(..)   , Interruptibility(..)@@ -32,6 +36,7 @@  import Control.Exception (Exception(..)) import Data.Kind (Type)+import Data.Typeable (Typeable) import Foreign.C.Types (CInt)  import qualified Data.List as L@@ -40,6 +45,18 @@  data Connectedness = Connected | Unconnected +-- | This is used as a phantom type to distinguish+-- between pinned and unpinned memory. The distinction+-- drawn here concerns the manner in which the user+-- requested the byte array. Byte arrays resulting+-- from newPinnedByteArray# are considered 'Pinned'+-- while those resulting from newByteArray# are+-- considered 'Unpinned'. Even if the runtime decides+-- to pin a byte array requested by newByteArray#+-- (e.g. because it is larger than 3KB), it is+-- considered 'Unpinned' by this classification.+data Pinnedness = Pinned | Unpinned+ data Family = Internet Version | Unix  data Version = V4 | V6@@ -47,6 +64,54 @@ data Interruptibility = Interruptible | Uninterruptible  data Forkedness = Forked | Unforked++data SocketException :: Type where+  -- | A limit on the number of open file descriptors has been reached.+  --   This could be the per-process limit or the system limit.+  --   (@EMFILE@ and @ENFILE@)+  SocketFileDescriptorLimit :: SocketException++deriving stock instance Show SocketException+deriving stock instance Eq SocketException+deriving anyclass instance Exception SocketException++-- | Recoverable exceptions that happen when establishing stream listeners+-- or opening datagram sockets.+--+-- ==== __Discussion__+--+-- The recoverable exceptions that we encounter with stream sockets (established+-- with @socket@-@bind@-@listen@) and datagram sockets (established with+-- @socket@-@bind@) are the exact same exceptions. Consequently, we reuse+-- the same type in both case.+data BindException :: Family -> Type where+  -- | The address is protected, and the user is not the superuser. This most+  --   commonly happens when trying to bind to a port below 1024. On Linux,+  --   When it is necessary to bind to such a port on Linux, consider using the+  --   <http://man7.org/linux/man-pages/man7/capabilities.7.html CAP_NET_BIND_SERVICE>+  --   capability instead of running the process as root. (@EACCES@)+  BindPermissionDenied :: BindException f+  -- | The given address is already in use. This can also happen if the+  --   address was recently bound to and closed. As mandated by+  --   <https://tools.ietf.org/html/rfc793 RFC 793>, such a socket remains+  --   in the @TIME-WAIT@ state for a while (commonly 30 to 120 seconds),+  --   and no new socket may be bound to the address during this period.+  --   (@EADDRINUSE@ with specified port)+  BindAddressInUse :: BindException f+  -- | The port number was specified as zero, but upon attempting to+  --   bind to an ephemeral port, it was determined that all port numbers+  --   numbers in the ephemeral port range are currently in use.+  --   (@EADDRINUSE@ with unspecified port)+  BindEphemeralPortsExhausted :: BindException ('Internet v)+  -- | A limit on the number of open file descriptors has been reached.+  --   This could be the per-process limit or the system limit.+  --   (@EMFILE@ and @ENFILE@)+  BindFileDescriptorLimit :: BindException f++deriving stock instance Show (BindException f)+deriving stock instance Eq (BindException f)+deriving anyclass instance Typeable f => Exception (BindException f)+  data SocketUnrecoverableException = SocketUnrecoverableException   { modules :: String
src-internal/Socket/Connected/UnmanagedBytes.hs view
@@ -15,7 +15,7 @@ import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes)) import Socket.AddrLength (Buffer,length,advance) import Foreign.C.Error (Errno)-import Foreign.C.Types (CInt,CSize)+import Foreign.C.Types (CSize) import System.Posix.Types (Fd) import qualified Posix.Socket as S
src-internal/Socket/Datagram.hs view
@@ -24,6 +24,8 @@   --   number of bytes that were successfully copied into the   --   send buffer. The datagram does still get sent when this   --   happens.+  --   TODO: Can this actually happen? SendMessageSize might be+  --   what actually happens in this case.   SendTruncated :: !Int -> SendException i   -- | Attempted to send to a broadcast address.   SendBroadcasted :: SendException i@@ -33,8 +35,16 @@   --   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.)+  --   it is how Linux does it.) This does not happen reliably.+  --   Certain network environment block the ICMP traffic needed+  --   for the sender to learn that its messages have not been+  --   delivered. This happens with UNIX-domain sockets nothing+  --   is bound to the peer address. In that context, this exception+  --   happens reliably. (@ECONNREFUSED@ or @ENOTCONN@)   SendConnectionRefused :: SendException i+  -- | The size of the message made it impossible for the message+  -- to be delivered atomically. (@EMSGSIZE@)+  SendMessageSize :: SendException i   -- | STM-style interrupt (much safer than C-style interrupt)   SendInterrupted :: SendException 'Interruptible
src-internal/Socket/Destined/IPv4/UnmanagedBytes.hs view
@@ -14,8 +14,7 @@  import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes)) import Foreign.C.Error (Errno)-import Foreign.C.Types (CInt,CSize)-import GHC.Exts (RealWorld)+import Foreign.C.Types (CSize) import Net.Types (IPv4(..)) import Socket.IPv4 (Peer(..)) import Socket.AddrLength (advance,length)@@ -37,9 +36,6 @@     { S.port = S.hostToNetworkShort port     , S.address = S.hostToNetworkLong (getIPv4 address)     }--intToCInt :: Int -> CInt-intToCInt = fromIntegral  intToCSize :: Int -> CSize intToCSize = fromIntegral
src-internal/Socket/Discard.hs view
@@ -14,17 +14,23 @@   ( PeerlessSlab(..)   , newPeerlessSlab   , freezePeerlessSlab+  , freezePeerlessSlabAsByteString+  , replenishPeerlessSlab+  , replenishPinnedPeerlessSlab   ) where  import Control.Monad.Primitive (primitive,primitive_)-import Data.Primitive (ByteArray(..))-import Data.Primitive.Unlifted.Array (MutableUnliftedArray,UnliftedArray)+import Data.ByteString (ByteString)+import Data.Kind (Type)+import Data.Primitive (ByteArray(..),SmallMutableArray,SmallArray) import Data.Primitive (MutablePrimArray,MutableByteArray)-import Data.Primitive (SmallArray,SmallMutableArray)-import Data.Word (Word16)+import Data.Primitive.Unlifted.Array (MutableUnliftedArray,UnliftedArray) import Foreign.C.Types (CInt) import GHC.Exts (RealWorld,Int(I#))+import Socket (Pinnedness(Pinned,Unpinned)) import Socket.Error (die)+import Socket.Interop (fromPinned)+import Socket.Slab (replenishPinned,replenishUnpinned)  import qualified GHC.Exts as Exts import qualified Data.Primitive as PM@@ -33,58 +39,112 @@ -- | A slab of memory for bulk datagram ingest (via @recvmmsg@). -- Slabs are not safe for concurrent access. This variant does -- not have space for the peer addresses.-data PeerlessSlab = PeerlessSlab-  { sizes :: !(MutablePrimArray RealWorld CInt)-    -- ^ Buffer for returned datagram lengths-  , payloads :: !(MutableUnliftedArray RealWorld (MutableByteArray RealWorld))-    -- ^ Buffers for datagram payloads, no slicing-  }+data PeerlessSlab :: Pinnedness -> Type where+  PeerlessSlab ::+    -- Invariant: payload sizes are nondecreasing. An+    -- implication is that zero-size payloads come first.+    -- Invariant: All non-zero-size payloads match the+    -- phantom pinnedness. All zero-size payloads are+    -- unpinned.+    { sizes :: !(MutablePrimArray RealWorld CInt)+      -- ^ Buffer for returned datagram lengths+    , payloads :: !(MutableUnliftedArray RealWorld (MutableByteArray RealWorld))+      -- ^ Buffers for datagram payloads, no slicing+    } -> PeerlessSlab p  -- | Allocate a slab that is used to receive multiple datagrams at -- the same time without receiving any information about the peer. newPeerlessSlab ::      Int -- ^ maximum datagrams-  -> Int -- ^ maximum size of individual datagram -  -> IO PeerlessSlab-newPeerlessSlab !n !m = if n >= 1 && m >= 1+  -> IO (PeerlessSlab p)+newPeerlessSlab !n = if n >= 1   then do     sizes <- PM.newPrimArray n-    payloads <- PM.unsafeNewUnliftedArray n-    let go !ix = if ix > (-1)-          then do-            writeMutableByteArrayArray payloads ix =<< PM.newByteArray m-            go (ix - 1)-          else pure ()-    go (n - 1)+    tombstone <- PM.newByteArray 0+    payloads <- PM.newUnliftedArray n tombstone     pure PeerlessSlab{sizes,payloads}-  else die "newSlab"+  else die "newPeerlessSlab"  -- | Freeze the specified number of messages in-place and return--- them Replaces all of the frozen messages with tombstones so that+-- them. Replaces all of the frozen messages with tombstones so that -- new buffers can be allocated before the next reception. End users -- should not need this function.-freezePeerlessSlab :: PeerlessSlab -> Int -> IO (UnliftedArray ByteArray)+freezePeerlessSlab ::+     PeerlessSlab p+  -> Int+  -> IO (UnliftedArray ByteArray) freezePeerlessSlab slab n = do   msgs <- PM.unsafeNewUnliftedArray n-  freezeSlabGo slab msgs (n - 1)+  tombstone <- PM.newByteArray 0+  freezeSlabGo slab tombstone msgs (n - 1) +errorThunk :: a+{-# noinline errorThunk #-}+errorThunk = error "Socket.Discard: uninitialized element"++freezePeerlessSlabAsByteString ::+     PeerlessSlab 'Pinned+  -> Int+  -> IO (SmallArray ByteString)+freezePeerlessSlabAsByteString slab n = do+  msgs <- PM.newSmallArray n errorThunk+  tombstone <- PM.newByteArray 0+  freezeSlabByteStringGo slab tombstone msgs (n - 1)++freezeSlabByteStringGo ::+     PeerlessSlab 'Pinned+  -> MutableByteArray RealWorld -- tombstone+  -> SmallMutableArray RealWorld ByteString+  -> Int+  -> IO (SmallArray ByteString)+freezeSlabByteStringGo slab@PeerlessSlab{payloads,sizes} !tombstone !arr !ix = if ix > (-1)+  then do+    !size <- PM.readPrimArray sizes ix+    -- Remove the byte array from the array of payloads, freeze it, and+    -- replace it with a tombstone.+    payloadMut <- readMutableByteArrayArray payloads ix+    writeMutableByteArrayArray payloads ix tombstone+    let sz = cintToInt size+    !payload <- PM.resizeMutableByteArray payloadMut sz+    PM.writeSmallArray arr ix $! fromPinned payload 0 sz+    freezeSlabByteStringGo slab tombstone arr (ix - 1)+  else PM.unsafeFreezeSmallArray arr+ freezeSlabGo ::-     PeerlessSlab+     PeerlessSlab p+  -> MutableByteArray RealWorld -- tombstone   -> MutableUnliftedArray RealWorld ByteArray   -> Int   -> IO (UnliftedArray ByteArray)-freezeSlabGo slab@PeerlessSlab{payloads,sizes} !arr !ix = if ix > (-1)+freezeSlabGo slab@PeerlessSlab{payloads,sizes} !tombstone !arr !ix = if ix > (-1)   then do     !size <- PM.readPrimArray sizes ix     -- Remove the byte array from the array of payloads, freeze it, and-    -- replace it with a freshly allocated byte array. +    -- replace it with a tombstone.     payloadMut <- readMutableByteArrayArray payloads ix-    originalSize <- PM.getSizeofMutableByteArray payloadMut-    writeMutableByteArrayArray payloads ix =<< PM.newByteArray originalSize+    writeMutableByteArrayArray payloads ix tombstone     !payload <- PM.unsafeFreezeByteArray =<< PM.resizeMutableByteArray payloadMut (cintToInt size)     writeByteArrayArray arr ix payload-    freezeSlabGo slab arr (ix - 1)+    freezeSlabGo slab tombstone arr (ix - 1)   else PM.unsafeFreezeUnliftedArray arr++-- | Ensure that every buffer can accomodate at least the+-- minimum number of bytes.+replenishPeerlessSlab ::+     PeerlessSlab 'Unpinned+  -> Int -- ^ Minimum Size+  -> IO () +replenishPeerlessSlab PeerlessSlab{payloads} minSz = do+  let sz = PM.sizeofMutableUnliftedArray payloads+  replenishUnpinned payloads minSz 0 sz++replenishPinnedPeerlessSlab ::+     PeerlessSlab 'Pinned+  -> Int -- ^ Minimum Size+  -> IO () +replenishPinnedPeerlessSlab PeerlessSlab{payloads} minSz = do+  let sz = PM.sizeofMutableUnliftedArray payloads+  replenishPinned payloads minSz 0 sz  cintToInt :: CInt -> Int cintToInt = fromIntegral
src-internal/Socket/EventManager.hs view
@@ -31,6 +31,7 @@ import Control.Applicative (liftA2,(<|>)) import Control.Concurrent (getNumCapabilities,forkOn,rtsSupportsBoundThreads) import Control.Concurrent.STM (TVar)+import Control.Exception (mask_) import Control.Monad (when) import Control.Monad.STM (atomically) import Data.Bits (countLeadingZeros,finiteBitSize,unsafeShiftL,(.|.),(.&.))@@ -158,8 +159,10 @@ data Manager = Manager   { variables :: !(MUArray (MUArray (TVar Token)))   , novars :: !(MUArray (TVar Token))-    -- An empty mutable array. This array is used to mark the absense of-    -- a tier-two array of TVars.+    -- This array is used to mark the absense of a tier-two array+    -- of TVars. It always has a single TVar in it. This TVar is+    -- used for synchonization when growing the variables list.+    -- Zero means unlocked. Anything else means locked.   , epoll :: !Fd   } @@ -168,7 +171,8 @@ manager = unsafePerformIO $ do   when (not rtsSupportsBoundThreads) $ do     fail $ "Socket.Event.manager: threaded runtime required"-  !novars <- PM.unsafeNewUnliftedArray 0+  !novars <- PM.unsafeNewUnliftedArray 1+  writeTVarArray novars 0 =<< STM.newTVarIO (Token 0)   !variables <- PM.unsafeNewUnliftedArray 32   let goX !ix = if ix >= 0         then do@@ -254,11 +258,26 @@               goVars (ix - 1)             else pure ()       goVars (len - 1)+      syncVar <- readTVarArray novars 0+      -- I do not feel great about the consistency guarantees concerning+      -- how TVars interacte with non-TVar memory.+      mask_ $ do+        -- Acquire the lock. This plays nicely with the way GHC allows+        -- retrying STM transactions to be interrupted in a mask.+        STM.atomically $ do+          Token n <- STM.readTVar syncVar+          case n of+            0 -> STM.writeTVar syncVar (Token 1)+            _ -> STM.retry+        varsTier2' <- readMutableUnliftedArrayArray variables ixTier1+        when (PM.sameMutableUnliftedArray varsTier2' novars) $ do+          writeMutableUnliftedArrayArray variables ixTier1 varsAttempt+        STM.atomically (STM.writeTVar syncVar (Token 0))       -- We ignore the success of casUnliftedArray. It does not actually       -- matter whether or not it succeeded. If it failed, some other       -- thread must have initialized the tier 2 arrays.-      (success,tier2) <- casMutableUnliftedArrayArray variables ixTier1 novars varsAttempt-      debug ("constructivelyLookupTier1: Created tier 2 array of length " ++ show len ++ " at index " ++ show ixTier1 ++ " with success " ++ show success)+      tier2 <- readMutableUnliftedArrayArray variables ixTier1+      debug ("constructivelyLookupTier1: Created tier 2 array of length " ++ show len ++ " at index " ++ show ixTier1)       pure (ixTier2,tier2)     else pure (ixTier2,varsTier2) @@ -485,7 +504,7 @@ --    something less that the full number of bytes requested, we want --    to unready the token. The next time we want to @recv@, we want to wait --    for the token to be ready before even attempting a @recv@.--- 2. We want to be properly notification when the peer shuts down. Epoll+-- 2. We want to be properly notified when the peer shuts down. Epoll --    reports this as EPOLLRDHUP. -- -- The difficulty is that, without persistent readiness, performing@@ -636,7 +655,7 @@ interruptibleWaitCounting :: TVar Int -> TVar Bool -> TVar Token -> IO Token interruptibleWaitCounting !counter !interrupt !tv = atomically $   -- We cannot go to the same lengths to avoid a transaction as-  -- we do in interruptibleWait. Notablely, the token check and+  -- we do in interruptibleWait. Notably, the token check and   -- the counter increment must happen in a transaction together.   ( do STM.check =<< STM.readTVar interrupt        pure interruptToken@@ -755,22 +774,3 @@ --  -- These interpretations suggest that logical disjunction will give  -- us the current readiness of the channel.--casMutableUnliftedArrayArray ::-     MutableUnliftedArray RealWorld (MutableUnliftedArray RealWorld a) -- ^ array-  -> Int -- ^ index-  -> (MutableUnliftedArray RealWorld a) -- ^ expected old value-  -> (MutableUnliftedArray RealWorld a) -- ^ new value-  -> IO (Bool,MutableUnliftedArray RealWorld a)-{-# INLINE casMutableUnliftedArrayArray #-}-casMutableUnliftedArrayArray (MutableUnliftedArray arr#) (I# i#) (MutableUnliftedArray old) (MutableUnliftedArray new) =-  -- All of this unsafeCoercing is really nasty business. This will go away-  -- once https://github.com/ghc-proposals/ghc-proposals/pull/203 happens.-  -- Also, this is unsound if the result is immidiately consumed by-  -- the FFI.-  IO $ \s0 ->-    let !uold = (unsafeCoerce# :: MutableArrayArray# RealWorld -> Any) old-        !unew = (unsafeCoerce# :: MutableArrayArray# RealWorld -> Any) new-     in case casArray# ((unsafeCoerce# :: MutableArrayArray# RealWorld -> MutableArray# RealWorld Any) arr#) i# uold unew s0 of-          (# s1, n, ur #) -> (# s1, (isTrue# (n ==# 0# ),MutableUnliftedArray ((unsafeCoerce# :: Any -> MutableArrayArray# RealWorld) ur)) #)-
src-internal/Socket/IPv4.hs view
@@ -18,6 +18,8 @@   , describeEndpoint   , freezeIPv4Slab   , newIPv4Slab+  , replenishIPv4Slab+  , replenishPinnedIPv4Slab   ) where  import Control.Exception (Exception)@@ -31,7 +33,9 @@ import Foreign.C.Types (CInt) import GHC.Exts (RealWorld,Int(I#)) import Net.Types (IPv4(..))+import Socket (Pinnedness(Pinned,Unpinned)) import Socket.Error (die)+import Socket.Slab (replenishPinned,replenishUnpinned)  import qualified Data.Primitive as PM import qualified Data.Primitive.Unlifted.Array as PM@@ -40,7 +44,7 @@ import qualified Net.IPv4 as IPv4 import qualified Posix.Socket as S --- | An peer for an IPv4 socket, connection, or listener.+-- | A peer for an IPv4 socket, connection, or listener. --   Everything is in host byte order, and the user is not --   responsible for performing any conversions. data Peer = Peer@@ -48,7 +52,9 @@   , port :: !Word16   } deriving stock (Eq,Show) --- | A message received from a peer.+-- | A message received from a peer. The payload may be pinned+-- or unpinned depending on the reception function that produced+-- the message. data Message = Message   { peer :: {-# UNPACK #-} !Peer   , payload :: !ByteArray@@ -57,32 +63,32 @@ -- | A slab of memory for bulk datagram ingest (via @recvmmsg@). This -- approach cuts down on allocations. Slabs are not safe for concurrent -- access. Do not share a slab across multiple threads.-data IPv4Slab = IPv4Slab-  { sizes :: !(MutablePrimArray RealWorld CInt)-    -- ^ Buffer for returned datagram lengths-  , peers :: !(MutablePrimArray RealWorld S.SocketAddressInternet)-    -- ^ Buffer for returned addresses-  , payloads :: !(MutableUnliftedArray RealWorld (MutableByteArray RealWorld))-    -- ^ Buffers for datagram payloads, no slicing-  }+data IPv4Slab :: Pinnedness -> Type where+  IPv4Slab ::+    -- Invariant: payload sizes are nondecreasing. An+    -- implication is that zero-size payloads come first.+    -- Invariant: All non-zero-size payloads match the+    -- phantom pinnedness. All zero-size payloads are+    -- unpinned.+    { sizes :: !(MutablePrimArray RealWorld CInt)+      -- ^ Buffer for returned datagram lengths+    , peers :: !(MutablePrimArray RealWorld S.SocketAddressInternet)+      -- ^ Buffer for returned addresses+    , payloads :: !(MutableUnliftedArray RealWorld (MutableByteArray RealWorld))+      -- ^ Buffers for datagram payloads, no slicing+    } -> IPv4Slab p  -- | Allocate a slab that is used to receive multiple datagrams at -- the same time, additionally storing the IPv4 addresses of the peers. newIPv4Slab ::      Int -- ^ maximum datagrams-  -> Int -- ^ maximum size of individual datagram -  -> IO IPv4Slab-newIPv4Slab !n !m = if n >= 1 && m >= 1+  -> IO (IPv4Slab p)+newIPv4Slab !n = if n >= 1   then do     sizes <- PM.newPrimArray n     peers <- PM.newPrimArray n-    payloads <- PM.unsafeNewUnliftedArray n-    let go !ix = if ix > (-1)-          then do-            writeMutableByteArrayArray payloads ix =<< PM.newByteArray m-            go (ix - 1)-          else pure ()-    go (n - 1)+    tombstone <- PM.newByteArray 0+    payloads <- PM.newUnliftedArray n tombstone     pure IPv4Slab{sizes,peers,payloads}   else die "newSlabIPv4" @@ -110,7 +116,12 @@   --   <http://man7.org/linux/man-pages/man7/capabilities.7.html CAP_NET_BIND_SERVICE>   --   capability instead of running the process as root. (@EACCES@)   SocketPermissionDenied :: SocketException-  -- | The given address is already in use. (@EADDRINUSE@ with specified port)+  -- | The given address is already in use. This can also happen if the+  --   address was recently bound to and closed. As mandated by+  --   <https://tools.ietf.org/html/rfc793 RFC 793>, such a socket remains+  --   in the @TIME-WAIT@ state for a while (commonly 30 to 120 seconds),+  --   and no new socket may be bound to the address during this period.+  --   (@EADDRINUSE@ with specified port)   SocketAddressInUse :: SocketException   -- | The port number was specified as zero, but upon attempting to   --   bind to an ephemeral port, it was determined that all port numbers@@ -130,29 +141,53 @@ -- messages with tombstones so that new buffers can be allocated -- before the next reception. End users should not need this function. freezeIPv4Slab ::-     IPv4Slab -- ^ The slab+     IPv4Slab p -- ^ The slab   -> Int -- ^ Number of messages in the slab.   -> IO (SmallArray Message) freezeIPv4Slab slab n = do   msgs <- PM.newSmallArray n errorThunk-  freezeSlabGo slab msgs (n - 1)+  tombstone <- PM.newByteArray 0+  freezeSlabGo slab tombstone msgs (n - 1) -freezeSlabGo :: IPv4Slab -> SmallMutableArray RealWorld Message -> Int -> IO (SmallArray Message)-freezeSlabGo slab@IPv4Slab{payloads,peers,sizes} !arr !ix = if ix > (-1)+freezeSlabGo ::+     IPv4Slab p+  -> MutableByteArray RealWorld -- tombstone+  -> SmallMutableArray RealWorld Message+  -> Int+  -> IO (SmallArray Message)+freezeSlabGo slab@IPv4Slab{payloads,peers,sizes} !tombstone !arr !ix = if ix > (-1)   then do     !size <- PM.readPrimArray sizes ix     !sockaddr <- PM.readPrimArray peers ix     -- Remove the byte array from the array of payloads, freeze it, and-    -- replace it with a freshly allocated byte array. +    -- replace it with a tombstone.     payloadMut <- readMutableByteArrayArray payloads ix-    originalSize <- PM.getSizeofMutableByteArray payloadMut-    !payload <- PM.unsafeFreezeByteArray =<< PM.resizeMutableByteArray payloadMut (cintToInt size)-    writeMutableByteArrayArray payloads ix =<< PM.newByteArray originalSize+    !payload <- PM.unsafeFreezeByteArray+      =<< PM.resizeMutableByteArray payloadMut (cintToInt size)+    writeMutableByteArrayArray payloads ix tombstone     let !peer = sockAddrToPeer sockaddr         !msg = Message {peer,payload}     PM.writeSmallArray arr ix msg-    freezeSlabGo slab arr (ix - 1)+    freezeSlabGo slab tombstone arr (ix - 1)   else PM.unsafeFreezeSmallArray arr++-- | Ensure that every buffer can accomodate at least the+-- minimum number of bytes.+replenishIPv4Slab ::+     IPv4Slab 'Unpinned+  -> Int -- ^ Minimum Size+  -> IO () +replenishIPv4Slab IPv4Slab{payloads} minSz = do+  let sz = PM.sizeofMutableUnliftedArray payloads+  replenishUnpinned payloads minSz 0 sz++replenishPinnedIPv4Slab ::+     IPv4Slab 'Pinned+  -> Int -- ^ Minimum Size+  -> IO () +replenishPinnedIPv4Slab IPv4Slab{payloads} minSz = do+  let sz = PM.sizeofMutableUnliftedArray payloads+  replenishPinned payloads minSz 0 sz  {-# NOINLINE errorThunk #-} errorThunk :: Message
src-internal/Socket/Interruptible.hs view
@@ -1,3 +1,4 @@+{-# language CPP #-} {-# language DataKinds #-}  module Socket.Interruptible@@ -9,19 +10,45 @@   , tokenToStreamReceiveException   , tokenToDatagramSendException   , tokenToDatagramReceiveException+  , tokenToSequencedPacketSendException+  , tokenToSequencedPacketReceiveException   ) where  import Socket (Interruptibility(Interruptible)) import Socket.EventManager (Token) import Control.Concurrent.STM (TVar)-import GHC.Exts (RuntimeRep(LiftedRep)) import qualified Socket.EventManager as EM import qualified Socket.Stream as Stream import qualified Socket.Datagram as Datagram+import qualified Socket.SequencedPacket as SequencedPacket -type InterruptRep = 'LiftedRep+#if MIN_VERSION_base(4,16,0)+import GHC.Exts (RuntimeRep(BoxedRep),Levity(Lifted))+#else+import GHC.Exts (RuntimeRep(LiftedRep))+#endif++type InterruptRep = +#if MIN_VERSION_base(4,16,0)+  'BoxedRep 'Lifted+#else+  'LiftedRep+#endif+ type Interrupt = TVar Bool type Intr = 'Interruptible++tokenToSequencedPacketSendException :: Token -> Either (SequencedPacket.SendException 'Interruptible) ()+{-# inline tokenToSequencedPacketSendException #-}+tokenToSequencedPacketSendException t = if EM.isInterrupt t+  then Left SequencedPacket.SendInterrupted+  else Right ()++tokenToSequencedPacketReceiveException :: Token -> Either (SequencedPacket.ReceiveException 'Interruptible) ()+{-# inline tokenToSequencedPacketReceiveException #-}+tokenToSequencedPacketReceiveException t = if EM.isInterrupt t+  then Left SequencedPacket.ReceiveInterrupted+  else Right ()  tokenToStreamSendException :: Token -> Int -> Either (Stream.SendException 'Interruptible) () {-# inline tokenToStreamSendException #-}
+ src-internal/Socket/SequencedPacket.hs view
@@ -0,0 +1,63 @@+{-# language DataKinds #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language GADTs #-}+{-# language KindSignatures #-}+{-# language StandaloneDeriving #-}++module Socket.SequencedPacket+  ( ReceiveException(..)+  , SendException(..)+  ) where++import Data.Kind (Type)+import Data.Typeable (Typeable)+import Control.Exception (Exception)+import Socket (Interruptibility(..))++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 :: ReceiveException i+  -- | The peer shutdown its writing channel. (zero-length chunk)+  ReceiveShutdown :: ReceiveException i+  -- | The peer reset the connection. (@ECONNRESET@)+  ReceiveReset :: ReceiveException i+  -- | STM-style interrupt (much safer than C-style interrupt)+  -- This provides the number of bytes received before the interrupt+  -- happened. For @receiveOnce@, this will always be zero, but+  -- for @receiveExactly@ and @receiveBetween@, it may be any+  -- non-negative number less than the number of bytes the caller+  -- intended to receive.+  ReceiveInterrupted :: ReceiveException 'Interruptible++deriving stock instance Eq (ReceiveException i)+deriving stock instance Show (ReceiveException i)+deriving anyclass instance Typeable i => Exception (ReceiveException i)++data SendException :: Interruptibility -> Type where+  -- | The datagram did not fit in the buffer. The field is the+  --   number of bytes that were successfully copied into the+  --   send buffer. The datagram does still get sent when this+  --   happens.+  SendTruncated :: !Int -> SendException i+  -- | The local socket has already shutdown its writing channel.+  --   Consequently, sending is no longer possible. This can happen+  --   even if the process does not @shutdown@ the socket. If the+  --   peer decides to @close@ the connection, the local operating system+  --   will shutdown both the reading and writing channels. (@EPIPE@)+  SendShutdown :: SendException i+  -- | The peer reset the connection.+  SendReset :: SendException i+  -- | STM-style interrupt (much safer than C-style interrupt).+  -- This provides the number of bytes sent before the interrupt+  -- happened. For @sendOnce@, this will always be zero, but+  -- for @send@, it may be any non-negative number less than the+  -- number of bytes the caller intended to send.+  SendInterrupted :: SendException 'Interruptible++deriving stock instance Eq (SendException i)+deriving stock instance Show (SendException i)+deriving anyclass instance Typeable i => Exception (SendException i)
+ src-internal/Socket/Slab.hs view
@@ -0,0 +1,51 @@+{-# language BangPatterns #-}++module Socket.Slab+  ( replenishUnpinned+  , replenishPinned+  ) where++import Data.Primitive (MutableByteArray)+import Data.Primitive.Unlifted.Array (MutableUnliftedArray)+import GHC.Exts (RealWorld)++import qualified Data.Primitive as PM+import qualified Data.Primitive.Unlifted.Array as PM++-- Notice that this only replaces buffers until we+-- reach a single buffer that is of sufficient size.+-- This works because the buffer sizes are nondecreasing.+replenishUnpinned ::+     MutableUnliftedArray RealWorld (MutableByteArray RealWorld)+  -> Int+  -> Int+  -> Int+  -> IO () +replenishUnpinned !xs !minSz !ix !sz = if ix < sz+  then do+    p <- PM.readUnliftedArray xs ix+    pSz <- PM.getSizeofMutableByteArray p+    if pSz < minSz+      then do+        PM.writeUnliftedArray xs ix =<< PM.newByteArray minSz+        replenishUnpinned xs minSz (ix + 1) sz+      else pure ()+  else pure ()++replenishPinned ::+     MutableUnliftedArray RealWorld (MutableByteArray RealWorld)+  -> Int+  -> Int+  -> Int+  -> IO () +replenishPinned !xs !minSz !ix !sz = if ix < sz+  then do+    p <- PM.readUnliftedArray xs ix+    pSz <- PM.getSizeofMutableByteArray p+    if pSz < minSz+      then do+        PM.writeUnliftedArray xs ix =<< PM.newPinnedByteArray minSz+        replenishPinned xs minSz (ix + 1) sz+      else pure ()+  else pure ()+
src-internal/Socket/Stream.hs view
@@ -78,7 +78,7 @@   ConnectHostUnreachable :: ConnectException ('Internet v) i   -- | All port numbers in the ephemeral port range are currently in   --   use. (@EADDRNOTAVAIL@).-  ConnectEphemeralPortsExhausted :: ConnectException d i+  ConnectEphemeralPortsExhausted :: ConnectException ('Internet v) i   -- | No one is listening on the remote address. (@ECONNREFUSED@)   ConnectRefused :: ConnectException d i   -- | Timeout while attempting connection. The server may be too busy@@ -91,6 +91,9 @@   ConnectTimeout :: ConnectException d i   -- | Remote socket does not match the local socket type. (@EPROTOTYPE@)   ConnectProtocolType :: ConnectException 'Unix i+  -- | The pathname used as the remote address in a call to @connect@+  --   did not exist. This only happens with UNIX-domain sockets. (@ENOENT@)+  ConnectNoEntry :: ConnectException 'Unix i   -- | STM-style interrupt (much safer than C-style interrupt)   ConnectInterrupted :: ConnectException d 'Interruptible
src-internal/Socket/Uninterruptible.hs view
@@ -7,6 +7,8 @@   , Interrupt   , Intr   , wait+  , tokenToSequencedPacketSendException+  , tokenToSequencedPacketReceiveException   , tokenToStreamSendException   , tokenToStreamReceiveException   , tokenToDatagramSendException@@ -21,12 +23,21 @@ import qualified Socket.EventManager as EM import qualified Socket.Stream as Stream import qualified Socket.Datagram as Datagram+import qualified Socket.SequencedPacket as SequencedPacket  -- We use Proxy# instead of () to ensure that this interrupt type -- has no runtime cost associated with it. type InterruptRep = 'TupleRep '[] type Interrupt = Proxy# Void type Intr = 'Uninterruptible++tokenToSequencedPacketSendException :: Token -> Either (SequencedPacket.SendException 'Uninterruptible) ()+{-# inline tokenToSequencedPacketSendException #-}+tokenToSequencedPacketSendException _ = Right ()++tokenToSequencedPacketReceiveException :: Token -> Either (SequencedPacket.ReceiveException 'Uninterruptible) ()+{-# inline tokenToSequencedPacketReceiveException #-}+tokenToSequencedPacketReceiveException _ = Right ()  tokenToStreamSendException ::      Token
src-interrupt/Socket/Interrupt.hsig view
@@ -10,6 +10,7 @@ import Socket (Interruptibility) import GHC.Exts (RuntimeRep,TYPE) import qualified Socket.Stream as Stream+import qualified Socket.SequencedPacket as SequencedPacket import qualified Socket.Datagram as Datagram  data InterruptRep :: RuntimeRep@@ -18,6 +19,12 @@  wait :: Interrupt -> TVar Token -> IO Token +tokenToSequencedPacketSendException ::+     Token+  -> Either (SequencedPacket.SendException Intr) ()+tokenToSequencedPacketReceiveException ::+     Token+  -> Either (SequencedPacket.ReceiveException Intr) () tokenToStreamSendException ::      Token   -> Int
src-no-mmsg/Socket/Datagram/Interruptible/MutableBytes/Many.hs view
@@ -35,7 +35,7 @@      -- @'Left' 'ReceiveInterrupted'@.   -> Socket c a      -- ^ Socket-  -> Socket.Discard.PeerlessSlab+  -> Socket.Discard.PeerlessSlab p      -- ^ Buffers into which sizes and payloads are received   -> IO (Either (ReceiveException 'Interruptible) Int) receiveMany intr (Socket fd) (Socket.Discard.PeerlessSlab{sizes,payloads}) =@@ -49,7 +49,7 @@      -- ^ Interrupt. On 'True', give up and return      -- @'Left' 'ReceiveInterrupted'@.   -> Socket 'Unconnected ('SCK.Internet 'SCK.V4) -- ^ Socket-  -> Socket.IPv4.IPv4Slab+  -> Socket.IPv4.IPv4Slab p      -- ^ Buffers into which sizes, addresses, and payloads      -- are received   -> IO (Either (ReceiveException 'Interruptible) Int)
src-no-mmsg/Socket/Datagram/Uninterruptible/MutableBytes/Many.hs view
@@ -11,7 +11,6 @@   ) where  import GHC.Exts (proxy#)-import GHC.IO (IO(..)) import Socket (Interruptibility(Uninterruptible)) import Socket (Connectedness(..)) import Socket.Datagram (Socket(..),ReceiveException)@@ -33,7 +32,7 @@ receiveMany ::       Socket c a      -- ^ Socket-  -> Socket.Discard.PeerlessSlab+  -> Socket.Discard.PeerlessSlab p      -- ^ Buffers into which sizes and payloads are received   -> IO (Either (ReceiveException 'Uninterruptible) Int) receiveMany (Socket fd) (Socket.Discard.PeerlessSlab{sizes,payloads}) =@@ -44,7 +43,7 @@ -- to the structure-of-arrays. receiveManyFromIPv4 ::       Socket 'Unconnected ('SCK.Internet 'SCK.V4) -- ^ Socket-  -> Socket.IPv4.IPv4Slab+  -> Socket.IPv4.IPv4Slab p      -- ^ Buffers into which sizes, addresses, and payloads      -- are received   -> IO (Either (ReceiveException 'Uninterruptible) Int)
+ src-stream-receive/SequencedPacket/Receive/Indefinite.hs view
@@ -0,0 +1,85 @@+{-# language BangPatterns #-}+{-# language LambdaCase #-}+{-# language MultiWayIf #-}++module SequencedPacket.Receive.Indefinite+  ( receive+  , receiveAttempt+  ) where++import Control.Concurrent.STM (TVar)+import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK, eCONNRESET)+import Foreign.C.Types (CSize)+import Socket.Error (die)+import Socket.EventManager (Token)+import Socket.SequencedPacket (ReceiveException(..))+import Socket.Buffer (Buffer)+import Socket.Interrupt (Interrupt,Intr,wait,tokenToSequencedPacketReceiveException)+import System.Posix.Types (Fd)++import qualified Foreign.C.Error.Describe as D+import qualified Socket.EventManager as EM+import qualified Socket.Buffer as Buffer+import qualified Stream.Receive as Receive+import qualified Linux.Socket as L++receive ::+     Interrupt+  -> Fd+  -> Buffer+  -> IO (Either (ReceiveException Intr) Int)+receive !intr !sock !buf = do+  let !mngr = EM.manager+  tv <- EM.reader mngr sock+  token0 <- wait intr tv+  case tokenToSequencedPacketReceiveException token0 of+    Left err -> pure (Left err)+    Right _ -> receiveLoop intr tv token0 sock buf++receiveAttempt ::+     Fd -- ^ Socket+  -> Buffer -- ^ Buffer+  -> IO (Either (ReceiveException Intr) (Maybe Int))+{-# inline receiveAttempt #-}+receiveAttempt !fd !buf = do+  e <- Receive.receiveOnce fd buf+  case e of+    Left err ->+      if | err == eWOULDBLOCK || err == eAGAIN -> pure (Right Nothing)+         | err == eCONNRESET -> pure (Left ReceiveReset)+         | otherwise -> die $ concat+             [ "Socket.SequencedPacket.receive: " +             , describeErrorCode err+             ]+    Right recvSz -> do+      let !recvSzInt = csizeToInt recvSz+      -- TODO: This is sort of a hack, but it mostly works. We should+      -- use MSG_TRUNC, but I am too lazy to fix this at the moment.+      if recvSzInt == Buffer.length buf+        then pure (Left ReceiveTruncated)+        else pure (Right (Just recvSzInt))++receiveLoop ::+     Interrupt+  -> TVar Token+  -> Token+  -> Fd -- ^ Socket+  -> Buffer -- ^ Buffer+  -> IO (Either (ReceiveException Intr) Int)+receiveLoop !intr !tv !token0 !fd !buf = receiveAttempt fd buf >>= \case+  Left err -> pure (Left err)+  Right m -> case m of+    Nothing -> do+      EM.unready token0 tv+      token1 <- wait intr tv+      case tokenToSequencedPacketReceiveException token1 of+        Left err -> pure (Left err)+        Right _ -> receiveLoop intr tv token1 fd buf+    Just !r -> pure (Right r)++csizeToInt :: CSize -> Int+csizeToInt = fromIntegral++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"+
+ src-stream-send/SequencedPacket/Send/Indefinite.hs view
@@ -0,0 +1,82 @@+{-# language BangPatterns #-}+{-# language LambdaCase #-}+{-# language MultiWayIf #-}++module SequencedPacket.Send.Indefinite+  ( send+  , attemptSend+  ) where++import Control.Concurrent.STM (TVar)+import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK)+import Foreign.C.Error (eCONNRESET,ePIPE)+import Foreign.C.Types (CSize)+import Socket.Error (die)+import Socket.EventManager (Token)+import Socket.SequencedPacket (SendException(..))+import Socket.Buffer (Buffer)+import Socket.Interrupt (Interrupt,Intr,wait,tokenToSequencedPacketSendException)+import System.Posix.Types (Fd)++import qualified Foreign.C.Error.Describe as D+import qualified Socket.EventManager as EM+import qualified Socket.Buffer as Buffer+import qualified Stream.Send as Send++-- Send the entirely of the buffer, making a single call to+-- 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+  -> Fd+  -> Buffer+  -> IO (Either (SendException Intr) ())+send !intr !sock !buf = do+  let !mngr = EM.manager+  tv <- EM.writer mngr sock+  token0 <- wait intr tv+  case tokenToSequencedPacketSendException token0 of+    Left err -> pure (Left err)+    Right _ -> sendLoop intr sock tv token0 buf++-- Never blocks.+attemptSend ::+     Fd+  -> Buffer+  -> IO (Either (SendException Intr) Bool)+attemptSend !sock !buf = Send.sendOnce sock buf >>= \case+  Left e ->+    if | e == eAGAIN || e == eWOULDBLOCK -> pure (Right False)+       | e == ePIPE -> pure (Left SendShutdown)+       | e == eCONNRESET -> pure (Left SendReset)+       | otherwise -> die ("Socket.SequencedPacket.send: " ++ describeErrorCode e)+  Right sz -> if csizeToInt sz == Buffer.length buf+    then pure $! Right True+    else pure $! Left $! SendTruncated $! csizeToInt sz++sendLoop ::+     Interrupt -> Fd -> TVar Token -> Token+  -> Buffer -> IO (Either (SendException Intr) ())+sendLoop !intr !sock !tv !old !buf =+  Send.sendOnce sock buf >>= \case+    Left e ->+      if | e == eAGAIN || e == eWOULDBLOCK -> do+             EM.unready old tv+             new <- wait intr tv+             case tokenToSequencedPacketSendException new of+               Left err -> pure (Left err)+               Right _ -> sendLoop intr sock tv new buf+         | e == ePIPE -> pure (Left SendShutdown)+         | e == eCONNRESET -> pure (Left SendReset)+         | otherwise -> die ("Socket.SequencedPacket.send: " ++ describeErrorCode e)+    Right sz -> if csizeToInt sz == Buffer.length buf+      then pure $! Right ()+      else pure $! Left $! SendTruncated $! csizeToInt sz++csizeToInt :: CSize -> Int+csizeToInt = fromIntegral++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"+
+ src/Socket/Datagram/Common.hs view
@@ -0,0 +1,25 @@+{-# language LambdaCase #-}++-- Functions like @close@ are the same regardless of whether+-- a socket is connected, unconnected, internet-domain, or+-- unix-domain.+module Socket.Datagram.Common+  ( close+  ) where++import Foreign.C.Error (Errno(..))+import Socket.Datagram (Socket(..))+import Socket.Error (die)+import qualified Posix.Socket as S+import qualified Foreign.C.Error.Describe as D++-- | Unbracketed function for closing a datagram socket. This must not+-- be used in conjunction with 'withSocket'.+close :: Socket c a -> IO ()+close (Socket fd) = S.uninterruptibleClose fd >>= \case+  Left err -> die ("Socket.Datagram.Common.close: " ++ describeErrorCode err)+  Right _ -> pure ()++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"+
src/Socket/Datagram/IPv4/Connected.hs view
@@ -19,6 +19,8 @@   , Message(..)     -- * Establish   , withSocket+  , open+  , close     -- * Exceptions   , SocketException(..)   , SD.ReceiveException(..)@@ -31,13 +33,13 @@ import Data.Word (Word16) import Foreign.C.Error (Errno(..),eACCES) import Foreign.C.Error (eNFILE,eMFILE,eADDRINUSE)-import GHC.IO (IO(..)) import Net.Types (IPv4(..)) import Socket (Connectedness(..),Family(..)) import Socket.Datagram (Socket(..)) import Socket.Datagram (SocketException(..)) import Socket.IPv4 (Peer(..),Message(..)) import Socket.Error (die)+import Socket.Datagram.Common (close)  import qualified Foreign.C.Error.Describe as D import qualified Linux.Socket as L@@ -46,17 +48,19 @@ import qualified Socket.EventManager as EM import qualified Socket.Datagram as SD -withSocket ::+-- | Unbracketed function for opening a socket. Be careful with+-- this function. Only call this in contexts where exceptions are+-- masked.+--+-- TODO: Rewrite this. Stop calling bind. Do not make the user+-- supply a local address.+open ::      Peer-     -- ^ Address and port to bind to   -> Peer-     -- ^ Peer address and port-  -> (Socket 'Connected ('SCK.Internet 'SCK.V4) -> Word16 -> IO a)-     -- ^ Callback providing the socket and the chosen port-  -> IO (Either SocketException a)-withSocket local@Peer{port = specifiedPort} !peer f = mask $ \restore -> do+  -> IO (Either SocketException (Socket 'Connected ('SCK.Internet 'SCK.V4), Word16))+open local@Peer{port = specifiedPort} !peer = do   -- TODO: This is mostly copied from the unconnected withSocket.-  e1 <- S.uninterruptibleSocket S.internet+  e1 <- S.uninterruptibleSocket S.Internet     (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)     S.defaultProtocol   case e1 of@@ -100,11 +104,23 @@               S.uninterruptibleConnect fd sockAddr >>= \case                 Left err ->                    die ("Socket.Datagram.IPv4.Connected.connect: " ++ describeErrorCode err)-                Right (_ :: ()) -> do-                  a <- onException (restore (f (Socket fd) actualPort)) (S.uninterruptibleErrorlessClose fd)-                  S.uninterruptibleClose fd >>= \case-                    Left err -> die ("Socket.Datagram.IPv4.Undestined.close: " ++ describeErrorCode err)-                    Right _ -> pure (Right a)+                Right (_ :: ()) -> pure (Right (Socket fd, actualPort))++withSocket ::+     Peer+     -- ^ Address and port to bind to+  -> Peer+     -- ^ Peer address and port+  -> (Socket 'Connected ('SCK.Internet 'SCK.V4) -> Word16 -> IO a)+     -- ^ Callback providing the socket and the chosen port+  -> IO (Either SocketException a)+withSocket !local !peer f = mask $ \restore -> open local peer >>= \case+  Left err -> pure (Left err)+  Right (Socket fd,actualPort) -> do+    a <- onException (restore (f (Socket fd) actualPort)) (S.uninterruptibleErrorlessClose fd)+    S.uninterruptibleClose fd >>= \case+      Left err -> die ("Socket.Datagram.IPv4.Undestined.close: " ++ describeErrorCode err)+      Right _ -> pure (Right a)  describeErrorCode :: Errno -> String describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"
src/Socket/Datagram/IPv4/Unconnected.hs view
@@ -6,6 +6,8 @@ {-# language LambdaCase #-} {-# language MagicHash #-} {-# language NamedFieldPuns #-}+{-# language ScopedTypeVariables #-}+{-# language TypeApplications #-} {-# language UnboxedTuples #-}  -- | Internet datagram sockets without a fixed destination.@@ -13,11 +15,16 @@   ( -- * Types     Socket(..)   , Family(..)+  , Version(..)   , Connectedness(..)   , Peer(..)   , Message(..)+  , SCK.Interruptibility(..)     -- * Establish   , withSocket+  , open+  , openOnDevice+  , close     -- * Exceptions   , SocketException(..)   ) where@@ -26,21 +33,96 @@ import Data.Word (Word16) import Foreign.C.Error (Errno(..),eACCES) import Foreign.C.Error (eNFILE,eMFILE,eADDRINUSE)-import GHC.IO (IO(..))+import Foreign.C.Types (CInt) import Net.Types (IPv4(..))-import Socket (Connectedness(..),Family(..))+import Socket (Connectedness(..),Family(..),Version(..)) import Socket.Datagram (Socket(..)) import Socket.Datagram (SocketException(..)) import Socket.Debug (debug) import Socket.IPv4 (Peer(..),Message(..),describeEndpoint) import Socket.Error (die)+import Socket.Datagram.Common (close)+import System.Posix.Types (Fd)  import qualified Foreign.C.Error.Describe as D import qualified Linux.Socket as L import qualified Posix.Socket as S+import qualified Data.Primitive as PM import qualified Socket as SCK import qualified Socket.EventManager as EM +openOnDevice ::+     Peer -- ^ Address and port to use+  -> PM.ByteArray -- ^ Local network interface name+  -> IO (Either SocketException (Socket 'Unconnected ('SCK.Internet 'SCK.V4), Word16))+openOnDevice !local !intf = do+  e1 <- S.uninterruptibleSocket S.Internet+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketException err+    Right fd -> do+      r <- S.uninterruptibleSetSocketOptionByteArray+        fd S.levelSocket S.bindToDevice+        intf (fromIntegral @Int @CInt (PM.sizeofByteArray intf))+      -- We throw a bogus error here. TODO: fix this.+      case r of+        Left _ -> do+          S.uninterruptibleErrorlessClose fd+          pure (Left SocketPermissionDenied)+        Right (_ :: ()) -> prepareSocket local fd++open ::+     Peer+     -- ^ Address and port to use+  -> IO (Either SocketException (Socket 'Unconnected ('SCK.Internet 'SCK.V4), Word16))+open !local = do+  e1 <- S.uninterruptibleSocket S.Internet+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketException err+    Right fd -> prepareSocket local fd++prepareSocket ::+     Peer+  -> Fd+  -> IO (Either SocketException (Socket 'Unconnected ('SCK.Internet 'SCK.V4), Word16))+prepareSocket !local@Peer{port = specifiedPort} !fd = do+  let !mngr = EM.manager+  EM.register mngr fd+  e2 <- S.uninterruptibleBind fd+    (S.encodeSocketAddressInternet (endpointToSocketAddressInternet local))+  debug ("withSocket: requested binding for " ++ describeEndpoint local)+  case e2 of+    Left err -> do+      -- We intentionally discard any exceptions thrown by close.+      -- There is simply nothing that can be done with them.+      S.uninterruptibleErrorlessClose fd+      handleBindException specifiedPort err+    Right _ -> do+      eactualPort <- if specifiedPort == 0+        then S.uninterruptibleGetSocketName fd S.sizeofSocketAddressInternet >>= \case+          Left err -> do+            S.uninterruptibleErrorlessClose fd+            die ("Socket.Datagram.IPv4.Undestined.getsockname: " ++ describeErrorCode err)+          Right (sockAddrRequiredSz,sockAddr) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet+            then case S.decodeSocketAddressInternet sockAddr of+              Just S.SocketAddressInternet{port = actualPort} -> do+                let cleanPort = S.networkToHostShort actualPort+                debug ("withSocket: successfully bound " ++ describeEndpoint local ++ " and got port " ++ show cleanPort)+                pure (Right cleanPort)+              Nothing -> do+                S.uninterruptibleErrorlessClose fd+                die "Socket.Datagram.IPv4.Unconnected: non-internet socket family"+            else do+              S.uninterruptibleErrorlessClose fd+              die "Socket.Datagram.IPv4.Unconnected: socket address size"+        else pure (Right specifiedPort)+      case eactualPort of+        Left err -> pure (Left err)+        Right actualPort -> pure (Right (Socket fd, actualPort))+ -- | Open a socket and run the supplied callback on it. This closes the socket -- when the callback finishes or when an exception is thrown. Do not return  -- the socket from the callback. This leads to undefined behavior. If the@@ -54,53 +136,13 @@   -> (Socket 'Unconnected ('SCK.Internet 'SCK.V4)  -> Word16 -> IO a)      -- ^ Callback providing the socket and the chosen port   -> IO (Either SocketException a)-withSocket endpoint@Peer{port = specifiedPort} f = mask $ \restore -> do-  debug ("withSocket: opening socket " ++ describeEndpoint endpoint)-  e1 <- S.uninterruptibleSocket S.internet-    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)-    S.defaultProtocol-  debug ("withSocket: opened socket " ++ describeEndpoint endpoint)-  case e1 of-    Left err -> handleSocketException err-    Right fd -> do-      let !mngr = EM.manager-      EM.register mngr fd-      e2 <- S.uninterruptibleBind fd-        (S.encodeSocketAddressInternet (endpointToSocketAddressInternet endpoint))-      debug ("withSocket: requested binding for " ++ describeEndpoint endpoint)-      case e2 of-        Left err -> do-          -- We intentionally discard any exceptions thrown by close.-          -- There is simply nothing that can be done with them.-          S.uninterruptibleErrorlessClose fd-          handleBindException specifiedPort err-        Right _ -> do-          eactualPort <- if specifiedPort == 0-            then S.uninterruptibleGetSocketName fd S.sizeofSocketAddressInternet >>= \case-              Left err -> do-                S.uninterruptibleErrorlessClose fd-                die ("Socket.Datagram.IPv4.Undestined.getsockname: " ++ describeErrorCode err)-              Right (sockAddrRequiredSz,sockAddr) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet-                then case S.decodeSocketAddressInternet sockAddr of-                  Just S.SocketAddressInternet{port = actualPort} -> do-                    let cleanPort = S.networkToHostShort actualPort-                    debug ("withSocket: successfully bound " ++ describeEndpoint endpoint ++ " and got port " ++ show cleanPort)-                    pure (Right cleanPort)-                  Nothing -> do-                    S.uninterruptibleErrorlessClose fd-                    die "Socket.Datagram.IPv4.Undestined: non-internet socket family"-                else do-                  S.uninterruptibleErrorlessClose fd-                  die "Socket.Datagram.IPv4.Undestined: socket address size"-            else pure (Right specifiedPort)-          case eactualPort of-            Left err -> pure (Left err)-            Right actualPort -> do-              a <- onException (restore (f (Socket fd) actualPort)) (S.uninterruptibleErrorlessClose fd)-              S.uninterruptibleClose fd >>= \case-                Left err -> die ("Socket.Datagram.IPv4.Undestined.close: " ++ describeErrorCode err)-                Right _ -> pure (Right a)-+withSocket local f = mask $ \restore -> open local >>= \case+  Left err -> pure (Left err)+  Right (Socket fd,actualPort) -> do+    a <- onException (restore (f (Socket fd) actualPort)) (S.uninterruptibleErrorlessClose fd)+    S.uninterruptibleClose fd >>= \case+      Left err -> die ("Socket.Datagram.IPv4.Unconnected.close: " ++ describeErrorCode err)+      Right _ -> pure (Right a)  endpointToSocketAddressInternet :: Peer -> S.SocketAddressInternet endpointToSocketAddressInternet (Peer {address, port}) = S.SocketAddressInternet@@ -122,7 +164,7 @@ handleBindException !thePort !e   | e == eACCES = pure (Left SocketPermissionDenied)   | e == eADDRINUSE = if thePort == 0-      then pure (Left SocketAddressInUse)-      else pure (Left SocketEphemeralPortsExhausted)+      then pure (Left SocketEphemeralPortsExhausted)+      else pure (Left SocketAddressInUse)   | otherwise = die       ("Socket.Datagram.IPv4.Undestined.bind: " ++ describeErrorCode e)
src/Socket/Datagram/Interruptible/ByteString.hs view
@@ -13,6 +13,10 @@     -- * Receive   , receive   , receiveFromIPv4+    -- * Types+  , Peer(..)+  , ReceiveException(..)+  , SendException(..)     -- * Slabs     -- ** Types   , PeerlessSlab(..)@@ -32,8 +36,8 @@ 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.Datagram (Socket(..),SendException(..),ReceiveException(..))+import Socket.IPv4 (Peer(..),newIPv4Slab,IPv4Slab(..)) import Socket.Discard (PeerlessSlab(..),newPeerlessSlab) import Socket.Interop (fromPinned)
src/Socket/Datagram/Interruptible/Bytes.hs view
@@ -13,6 +13,19 @@     -- * Receive Many   , receiveMany   , receiveManyFromIPv4+  , receiveManyPinnedFromIPv4+    -- * Types+  , Pinnedness(..)+  , Message(..)+  , Peer(..)+  , ReceiveException(..)+    -- * Slabs+    -- ** Types+  , PeerlessSlab(..)+  , IPv4Slab(..)+    -- ** Functions+  , newPeerlessSlab+  , newIPv4Slab   ) where  import Control.Concurrent.STM (TVar)@@ -20,13 +33,16 @@ import Data.Primitive (ByteArray,SmallArray) import Data.Primitive.Unlifted.Array (UnliftedArray) import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset(..))-import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Interruptible))+import Socket (Connectedness(..),Family(..),Version(..))+import Socket (Pinnedness(Pinned,Unpinned))+import Socket (Interruptibility(Interruptible)) import Socket.Address (posixToIPv4Peer)-import Socket.Datagram (Socket(..),ReceiveException)-import Socket.IPv4 (Message(..),IPv4Slab(..))+import Socket.Datagram (Socket(..),ReceiveException(..))+import Socket.IPv4 (Peer(..),Message(..),IPv4Slab(..),freezeIPv4Slab)+import Socket.IPv4 (newIPv4Slab,replenishIPv4Slab,replenishPinnedIPv4Slab)+import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)  import qualified Data.Primitive as PM-import qualified Socket.IPv4 import qualified Socket.Discard import qualified Socket as SCK import qualified Socket.Datagram.Interruptible.MutableBytes.Many as MM@@ -73,23 +89,43 @@      -- ^ Interrupt. On 'True', give up and return      -- @'Left' 'ReceiveInterrupted'@.   -> Socket 'Unconnected ('SCK.Internet 'SCK.V4) -- ^ Socket-  -> Socket.IPv4.IPv4Slab -- ^ Buffers for reception+  -> Socket.IPv4.IPv4Slab 'Unpinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram   -> IO (Either (ReceiveException 'Interruptible) (SmallArray Message))-receiveManyFromIPv4 intr sock slab = do+receiveManyFromIPv4 intr sock slab maxSz = do+  replenishIPv4Slab slab maxSz   MM.receiveManyFromIPv4 intr sock slab >>= \case     Left err -> pure (Left err)     Right n -> do       arr <- Socket.IPv4.freezeIPv4Slab slab n       pure (Right arr) +receiveManyPinnedFromIPv4 ::+     TVar Bool+     -- ^ Interrupt. On 'True', give up and return+     -- @'Left' 'ReceiveInterrupted'@.+  -> Socket 'Unconnected ('SCK.Internet 'SCK.V4) -- ^ Socket+  -> Socket.IPv4.IPv4Slab 'Pinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram+  -> IO (Either (ReceiveException 'Interruptible) (SmallArray Message))+receiveManyPinnedFromIPv4 intr sock slab maxSz = do+  replenishPinnedIPv4Slab slab maxSz+  MM.receiveManyFromIPv4 intr sock slab >>= \case+    Left err -> pure (Left err)+    Right n -> do+      arr <- Socket.IPv4.freezeIPv4Slab slab n+      pure (Right arr)+ receiveMany ::      TVar Bool      -- ^ Interrupt. On 'True', give up and return      -- @'Left' 'ReceiveInterrupted'@.   -> Socket 'Unconnected ('SCK.Internet 'SCK.V4) -- ^ Socket-  -> Socket.Discard.PeerlessSlab -- ^ Buffers for reception+  -> PeerlessSlab 'Unpinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram   -> IO (Either (ReceiveException 'Interruptible) (UnliftedArray ByteArray))-receiveMany intr sock slab = do+receiveMany intr sock slab maxSz = do+  Socket.Discard.replenishPeerlessSlab slab maxSz   MM.receiveMany intr sock slab >>= \case     Left err -> pure (Left err)     Right n -> do
src/Socket/Datagram/Uninterruptible/ByteString.hs view
@@ -13,6 +13,8 @@     -- * Receive   , receive   , receiveFromIPv4+    -- * Receive Many+  , receiveMany     -- * Slabs     -- ** Types   , PeerlessSlab(..)@@ -26,21 +28,25 @@ import Data.ByteString.Unsafe (unsafeUseAsCStringLen) import Data.Bytes.Types (UnmanagedBytes(UnmanagedBytes)) import Data.Bytes.Types (MutableBytes(MutableBytes))+import Data.Primitive (SmallArray) 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 (Pinnedness(Pinned)) import Socket.Datagram (Socket(..),SendException,ReceiveException) import Socket.IPv4 (Peer,newIPv4Slab,IPv4Slab(..)) import Socket.Discard (PeerlessSlab(..),newPeerlessSlab) import Socket.Interop (fromPinned) +import qualified Socket.Discard 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+import qualified Socket.Datagram.Uninterruptible.MutableBytes.Many as MM  -- | Send a datagram using a socket with a pre-designated peer. This -- refers to a datagram socket for which POSIX @connect@ has locked@@ -88,3 +94,26 @@   V4R.receive proxy# sock (MutableBytes marr 0 len) addr >>= \case     Left err -> pure (Left err)     Right sz -> pure $! Right $! fromPinned marr 0 sz++-- | Receive multiple datagrams at the same time. This uses @recvmmsg@+-- on platforms that support it. This function is provided as a convenience+-- for users of the @bytestring@ library. A slightly more allocation-friendly+-- approach to receiving datagrams as bytestrings is to make use of+-- 'Socket.Datagram.Uninterruptible.Bytes.receiveManyPinned' from+-- @Socket.Datagram.Uninterruptible.Bytes@. Performing the conversion+-- from pinned 'ByteArray' to 'ByteString' while folding over the result+-- array may lead to fewer allocations depending on how the 'ByteString's+-- are used.+receiveMany ::+     Socket c a -- ^ Socket+  -> PeerlessSlab 'Pinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram+  -> IO (Either (ReceiveException 'Uninterruptible) (SmallArray ByteString))+receiveMany sock slab maxSz = do+  Socket.Discard.replenishPinnedPeerlessSlab slab maxSz+  MM.receiveMany sock slab >>= \case+    Left err -> pure (Left err)+    Right n -> do+      arr <- Socket.Discard.freezePeerlessSlabAsByteString slab n+      pure (Right arr)+
src/Socket/Datagram/Uninterruptible/Bytes.hs view
@@ -11,16 +11,22 @@   ( -- * Send     send   , sendToIPv4+  , trySend     -- * Receive   , receive   , receiveFromIPv4+  , tryReceive     -- * Receive Many   , receiveMany+  , receiveManyPinned   , receiveManyFromIPv4+  , receiveManyPinnedFromIPv4     -- * Types+  , Pinnedness(..)   , Message(..)   , Peer(..)   , ReceiveException(..)+  , SendException(..)     -- * Slabs     -- ** Types   , PeerlessSlab(..)@@ -36,10 +42,11 @@ import Data.Primitive.PrimArray.Offset (MutablePrimArrayOffset(..)) import GHC.Exts (proxy#) import Socket (Connectedness(..),Family(..),Version(..),Interruptibility(Uninterruptible))+import Socket (Pinnedness(Pinned,Unpinned)) import Socket.Address (posixToIPv4Peer)-import Socket.Datagram (Socket(..),SendException,ReceiveException(..))+import Socket.Datagram (Socket(..),SendException(..),ReceiveException(..)) import Socket.IPv4 (Peer(..),Message(..),IPv4Slab(..),freezeIPv4Slab)-import Socket.IPv4 (newIPv4Slab)+import Socket.IPv4 (newIPv4Slab,replenishIPv4Slab,replenishPinnedIPv4Slab) import Socket.Discard (PeerlessSlab(..),newPeerlessSlab)  import qualified Data.Primitive as PM@@ -60,6 +67,15 @@ send (Socket !sock) !buf =   CS.send proxy# () sock buf +-- | Variant of 'send' that never blocks. If the datagram cannot+-- immidiately be copied to the send queue, returns @False@.+trySend ::+     Socket 'Connected a -- ^ Socket with designated peer+  -> Bytes -- ^ Slice of a buffer+  -> IO (Either (SendException 'Uninterruptible) Bool)+trySend (Socket !sock) !buf =+  CS.attemptSend () sock buf+ sendToIPv4 ::      Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer   -> Peer -- ^ Destination@@ -83,6 +99,22 @@       pure (Right r)     Left err -> pure (Left err) +-- | Variant of 'receive' that never blocks. If a datagram is not+-- immidiately available on the receive queue, returns @Nothing@.+tryReceive :: +     Socket c a -- ^ Socket+  -> Int -- ^ Maximum datagram size+  -> IO (Either (ReceiveException 'Uninterruptible) (Maybe ByteArray))+tryReceive (Socket !sock) !maxSz = do+  buf <- PM.newByteArray maxSz+  CR.receiveAttempt sock (MutableBytes buf 0 maxSz) () >>= \case+    Right msz -> case msz of+      Nothing -> pure (Right Nothing)+      Just sz -> do+        r <- PM.resizeMutableByteArray buf sz >>= PM.unsafeFreezeByteArray+        pure (Right (Just r))+    Left err -> pure (Left err)+ receiveFromIPv4 ::      Socket 'Unconnected ('Internet 'V4) -- ^ IPv4 socket without designated peer   -> Int -- ^ Maximum datagram size@@ -97,22 +129,65 @@       pure (Right (Message (posixToIPv4Peer posixAddr) r))     Left err -> pure (Left err) +-- | Receive multiple datagrams at the same time, returning the peers'+-- addresses as well. This uses @recvmmsg@ on platforms that support it. receiveManyFromIPv4 ::      Socket 'Unconnected ('Internet 'V4) -- ^ Socket-  -> Socket.IPv4.IPv4Slab -- ^ Buffers for reception+  -> Socket.IPv4.IPv4Slab 'Unpinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram   -> IO (Either (ReceiveException 'Uninterruptible) (SmallArray Message))-receiveManyFromIPv4 sock slab = do+receiveManyFromIPv4 sock slab maxSz = do+  replenishIPv4Slab slab maxSz   MM.receiveManyFromIPv4 sock slab >>= \case     Left err -> pure (Left err)     Right n -> do       arr <- Socket.IPv4.freezeIPv4Slab slab n       pure (Right arr) -receiveMany ::+-- | Receive multiple datagrams at the same time, returning the peers'+-- addresses as well. This uses @recvmmsg@ on platforms that support it.+-- The byte arrays in the resulting 'Message's are pinned.+receiveManyPinnedFromIPv4 ::      Socket 'Unconnected ('Internet 'V4) -- ^ Socket-  -> Socket.Discard.PeerlessSlab -- ^ Buffers for reception+  -> Socket.IPv4.IPv4Slab 'Pinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram+  -> IO (Either (ReceiveException 'Uninterruptible) (SmallArray Message))+receiveManyPinnedFromIPv4 sock slab maxSz = do+  replenishPinnedIPv4Slab slab maxSz+  MM.receiveManyFromIPv4 sock slab >>= \case+    Left err -> pure (Left err)+    Right n -> do+      arr <- Socket.IPv4.freezeIPv4Slab slab n+      pure (Right arr)++-- | Receive multiple datagrams at the same time. This uses @recvmmsg@+-- on platforms that support it.+receiveMany ::+     Socket c a -- ^ Socket+  -> Socket.Discard.PeerlessSlab 'Unpinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram   -> IO (Either (ReceiveException 'Uninterruptible) (UnliftedArray ByteArray))-receiveMany sock slab = do+receiveMany sock slab maxSz = do+  Socket.Discard.replenishPeerlessSlab slab maxSz+  -- TODO: use maxSz in receiveMany. Not terribly+  -- important for the time being.+  MM.receiveMany sock slab >>= \case+    Left err -> pure (Left err)+    Right n -> do+      arr <- Socket.Discard.freezePeerlessSlab slab n+      pure (Right arr)++-- | Receive multiple datagrams at the same time. This uses @recvmmsg@+-- on platforms that support it. All resulting byte arrays are pinned.+receiveManyPinned ::+     Socket c a -- ^ Socket+  -> Socket.Discard.PeerlessSlab 'Pinned -- ^ Buffers for reception+  -> Int -- ^ Maximum size of single datagram+  -> IO (Either (ReceiveException 'Uninterruptible) (UnliftedArray ByteArray))+receiveManyPinned sock slab maxSz = do+  Socket.Discard.replenishPinnedPeerlessSlab slab maxSz+  -- TODO: use maxSz in receiveMany. Not terribly+  -- important for the time being.   MM.receiveMany sock slab >>= \case     Left err -> pure (Left err)     Right n -> do
src/Socket/Datagram/Uninterruptible/MutableBytes.hs view
@@ -77,4 +77,3 @@   -> IO (Either (ReceiveException 'Uninterruptible) Int) receiveFromIPv4 (Socket !sock) !buf !addr =   V4R.receive proxy# sock buf addr-
+ src/Socket/Datagram/Unix/Connected.hs view
@@ -0,0 +1,162 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language NamedFieldPuns #-}+{-# language UnboxedTuples #-}+{-# language ScopedTypeVariables #-}+{-# language TypeApplications #-}++-- | Unix-domain datagram sockets with a fixed destination.+module Socket.Datagram.Unix.Connected+  ( -- * Types+    Socket(..)+  , Family(..)+  , Connectedness(..)+  , UnixAddress(..)+  , Message(..)+    -- * Establish+  , withSocket+  , withPair+  , connect+  , open+  , openPair+  , close+    -- * Exceptions+  , SocketException(..)+  , ConnectException(..)+  , SD.ReceiveException(..)+  , SD.SendException(..)+    -- * Examples+    -- $examples+  ) where++import Control.Exception (mask,onException)+import Data.Coerce (coerce)+import Data.Primitive (ByteArray)+import Foreign.C.Error (Errno(..),eNOENT,ePROTOTYPE)+import Foreign.C.Error (eNFILE,eMFILE,eCONNREFUSED)+import Socket (Connectedness(..),Family(..))+import Socket.Datagram (Socket(..))+import Socket (SocketException(..))+import Socket.Stream (ConnectException(..))+import Socket.Error (die)+import Socket.IPv4 (Message(..))+import Socket.Datagram.Common (close)+import Socket (Interruptibility(Uninterruptible))+import System.Posix.Types (Fd)++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 Socket.Datagram as SD+import qualified Socket.EventManager as EM+import qualified Socket.Pair as Pair++newtype UnixAddress = UnixAddress ByteArray++-- | Unbracketed function for opening a socket. Be careful with+-- this function. The resulting socket is a UNIX-domain datagram socket+-- on which @connect@ was called without first calling @bind@. This means+-- that it corresponds neither to an entry in the filesystem nor an entry+-- in the abstract socket namespace.+open ::+     UnixAddress+  -> IO (Either (ConnectException 'Unix 'Uninterruptible) (Socket 'Connected 'SCK.Unix))+open (UnixAddress remote) = do+  -- TODO: This is somewhat copied from the internet-domain+  -- socket code+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketException err+    Right fd -> do+      let !mngr = EM.manager+      EM.register mngr fd+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ remote+      S.uninterruptibleConnect fd sockAddr >>= \case+        Left err -> handleConnectException err+        Right (_ :: ()) -> pure (Right (Socket fd))++-- | Unbracketed function for opening a connected socket pair. All warnings+-- that apply to 'open' apply to this function as well.+openPair :: IO (Either SocketException (Socket 'Connected 'SCK.Unix, Socket 'Connected 'SCK.Unix))+openPair = coerce+  @(IO (Either SocketException (Fd,Fd)))+  @(IO (Either SocketException (Socket 'Connected 'SCK.Unix, Socket 'Connected 'SCK.Unix)))+  (Pair.open S.sequencedPacket)++-- | Run a callback that requires a pair of connected+-- datagram sockets. The sockets will be closed when the+-- callback completes.+withPair ::+     (Socket 'Connected 'SCK.Unix -> Socket 'Connected 'SCK.Unix -> IO a)+     -- ^ Callback providing the connected datagram sockets+  -> IO (Either SocketException a)+withPair f = mask $ \restore -> openPair >>= \case+  Left err -> pure (Left err)+  Right (Socket fdA, Socket fdB) -> do+    a <- onException+      (restore (f (Socket fdA) (Socket fdB)))+      (S.uninterruptibleErrorlessClose fdA *> S.uninterruptibleErrorlessClose fdB)+    S.uninterruptibleClose fdA >>= \case+      Left err -> die ("Socket.Datagram.Unix.Connected.close: " ++ describeErrorCode err)+      Right _ -> S.uninterruptibleClose fdB >>= \case+        Left err -> die ("Socket.Datagram.Unix.Connected.close: " ++ describeErrorCode err)+        Right _ -> pure (Right a)++withSocket ::+     UnixAddress+     -- ^ Peer address (to connect to)+  -> (Socket 'Connected 'SCK.Unix -> IO a)+     -- ^ Callback providing the socket and the chosen port+  -> IO (Either (ConnectException 'Unix 'Uninterruptible) a)+withSocket !peer f = mask $ \restore -> open peer >>= \case+  Left err -> pure (Left err)+  Right (Socket fd) -> do+    a <- onException (restore (f (Socket fd))) (S.uninterruptibleErrorlessClose fd)+    S.uninterruptibleClose fd >>= \case+      Left err -> die ("Socket.Datagram.Unix.Connected.close: " ++ describeErrorCode err)+      Right _ -> pure (Right a)++-- | Set the endpoint to connect to.+connect ::+     UnixAddress+     -- ^ Peer address+  -> Socket 'Connected 'SCK.Unix+     -- ^ Unix-domain datagram socket+  -> IO (Either (ConnectException 'Unix 'Uninterruptible) ())+connect (UnixAddress remote) (Socket fd) = do+  let sockAddr = id+        $ S.encodeSocketAddressUnix+        $ S.SocketAddressUnix+        $ remote+  S.uninterruptibleConnect fd sockAddr >>= \case+    Left err -> handleConnectException err+    Right (_ :: ()) -> pure (Right ())++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"++handleSocketException :: Errno -> IO (Either (ConnectException 'Unix i) a)+handleSocketException e+  | e == eMFILE = pure (Left ConnectFileDescriptorLimit)+  | e == eNFILE = pure (Left ConnectFileDescriptorLimit)+  | otherwise = die+      ("Socket.Datagram.Unix.Connected.socket: " ++ describeErrorCode e)++handleConnectException :: Errno -> IO (Either (ConnectException 'Unix i)  a)+handleConnectException e+  | e == eNOENT = pure (Left ConnectNoEntry)+  | e == ePROTOTYPE = pure (Left ConnectProtocolType)+  | e == eCONNREFUSED = pure (Left ConnectRefused)+  | otherwise = die+      ("Socket.Datagram.Unix.Connected.connect: " ++ describeErrorCode e)
+ src/Socket/Datagram/Unix/Unconnected.hs view
@@ -0,0 +1,107 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language NamedFieldPuns #-}+{-# language UnboxedTuples #-}++-- | Internet datagram sockets without a fixed destination.+module Socket.Datagram.Unix.Unconnected+  ( -- * Types+    Socket(..)+  , Family(..)+  , Connectedness(..)+  , UnixAddress(..)+  , Message(..)+    -- * Establish+  , withSocket+  , open+  , close+    -- * Exceptions+  , SocketException(..)+  ) where++import Control.Exception (mask,onException)+import Foreign.C.Error (Errno(..),eACCES)+import Foreign.C.Error (eNFILE,eMFILE,eADDRINUSE)+import Socket (Connectedness(..),Family(..))+import Socket.Datagram (Socket(..))+import Socket.Datagram (SocketException(..))+import Socket.IPv4 (Message(..))+import Socket.Error (die)+import Socket.Datagram.Common (close)+import Socket.Datagram.Unix.Connected (UnixAddress(..))++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 Socket.EventManager as EM++open ::+     UnixAddress+     -- ^ Address and port to use+  -> IO (Either SocketException (Socket 'Unconnected 'Unix))+open (UnixAddress path) = do+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketException err+    Right fd -> do+      let !mngr = EM.manager+      EM.register mngr fd+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ path+      e2 <- S.uninterruptibleBind fd sockAddr+      case e2 of+        Left err -> do+          -- We intentionally discard any exceptions thrown by close.+          -- There is simply nothing that can be done with them.+          S.uninterruptibleErrorlessClose fd+          handleBindException err+        Right _ -> pure (Right (Socket fd))++-- | Open a socket and run the supplied callback on it. This closes the socket+-- when the callback finishes or when an exception is thrown. Do not return +-- the socket from the callback. This leads to undefined behavior. If the+-- address @0.0.0.0@ is used, the socket receives on all network interfaces.+-- If the port 0 is used, an unused port is chosen by the operating system.+-- The callback provides the chosen port (or if the user specified a non-zero+-- port, the chosen port will be that value).+withSocket ::+     UnixAddress+     -- ^ Address and port to use+  -> (Socket 'Unconnected ('SCK.Internet 'SCK.V4) -> IO a)+     -- ^ Callback providing the socket and the chosen port+  -> IO (Either SocketException a)+withSocket local f = mask $ \restore -> open local >>= \case+  Left err -> pure (Left err)+  Right (Socket fd) -> do+    a <- onException (restore (f (Socket fd))) (S.uninterruptibleErrorlessClose fd)+    S.uninterruptibleClose fd >>= \case+      Left err -> die ("Socket.Datagram.IPv4.Unconnected.close: " ++ describeErrorCode err)+      Right _ -> pure (Right a)++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"++handleSocketException :: Errno -> IO (Either SocketException a)+handleSocketException e+  | e == eMFILE = pure (Left SocketFileDescriptorLimit)+  | e == eNFILE = pure (Left SocketFileDescriptorLimit)+  | otherwise = die+      ("Socket.Datagram.IPv4.Undestined.socket: " ++ describeErrorCode e)++handleBindException :: Errno -> IO (Either SocketException a)+handleBindException !e+  | e == eACCES = pure (Left SocketPermissionDenied)+  | e == eADDRINUSE = pure (Left SocketAddressInUse)+  | otherwise = die+      ("Socket.Datagram.IPv4.Undestined.bind: " ++ describeErrorCode e)+
+ src/Socket/Pair.hs view
@@ -0,0 +1,42 @@+{-# language BangPatterns #-}++module Socket.Pair+  ( open+  ) where++import Foreign.C.Error (Errno(..))+import Foreign.C.Error (eNFILE,eMFILE)+import Socket (SocketException(..))+import Socket.Error (die)+import System.Posix.Types (Fd)++import qualified Foreign.C.Error.Describe as D+import qualified Linux.Socket as L+import qualified Posix.Socket as S+import qualified Socket.EventManager as EM++-- Helper for opening a connected unix-domain socket pair. This is+-- used by Socket.Datagram.Unix.Connected and by Socket.SequencedPacket.Unix.+-- It could be used by Socket.Stream.Unix if that module is ever written.+open :: S.Type -> IO (Either SocketException (Fd, Fd))+open !typ = do+  e1 <- S.uninterruptibleSocketPair S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) typ)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketException err+    Right (fdA,fdB) -> do+      let !mngr = EM.manager+      EM.register mngr fdA+      EM.register mngr fdB+      pure (Right (fdA, fdB))++handleSocketException :: Errno -> IO (Either SocketException a)+handleSocketException e+  | e == eMFILE = pure (Left SocketFileDescriptorLimit)+  | e == eNFILE = pure (Left SocketFileDescriptorLimit)+  | otherwise = die+      ("Socket.Pair.open: " ++ describeErrorCode e)++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"
+ src/Socket/SequencedPacket/Uninterruptible/Bytes.hs view
@@ -0,0 +1,81 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}++module Socket.SequencedPacket.Uninterruptible.Bytes+  ( send+  , trySend+  , receive+  , tryReceive+  ) where++import Socket.SequencedPacket.Unix (Connection(..))+import Data.Primitive (ByteArray)+import Data.Bytes.Types (Bytes,MutableBytes(MutableBytes))+import GHC.Exts (proxy#)+import Socket (Interruptibility(Uninterruptible))+import Socket.SequencedPacket (SendException,ReceiveException)+import Socket.SequencedPacket.Unix (Connection(..))++import qualified Data.Primitive as PM+import qualified Socket.SequencedPacket.Uninterruptible.Bytes.Send as Send+import qualified Socket.SequencedPacket.Uninterruptible.MutableBytes.Receive as Receive++-- Sadly, SEQPACKET sockets are poorly documented and, in Linux,+-- behave inconsisently depending on whether the socket is+-- internet-domain or unix-domain. More thorough coverage of+-- this is found at https://stackoverflow.com/q/3595684+--+-- Since internet-domain SEQPACKET sockets are rare and unsupported+-- by most hardware, we only support unix-domain sockets. On Linux,+-- we can do this by just reusing the send and receive functions for+-- datagram sockets.++-- | Send a message over a connection.+send ::+     Connection -- ^ Connection+  -> Bytes -- ^ Slice of a buffer+  -> IO (Either (SendException 'Uninterruptible) ())+send (Connection !sock) !buf = Send.send proxy# sock buf++-- | Variant of 'send' that never blocks. If the datagram cannot+-- immidiately be copied to the send queue, returns @False@.+trySend ::+     Connection -- ^ Connection+  -> Bytes -- ^ Slice of a buffer+  -> IO (Either (SendException 'Uninterruptible) Bool)+trySend (Connection !sock) !buf = Send.attemptSend sock buf++-- | Receive a message.+receive ::+     Connection -- ^ Connection+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Uninterruptible) ByteArray)+{-# inline receive #-}+receive (Connection conn) !n = do+  !marr0 <- PM.newByteArray n+  Receive.receive proxy# conn (MutableBytes marr0 0 n) >>= \case+    Left err -> pure (Left err)+    Right sz -> do+      marr1 <- PM.resizeMutableByteArray marr0 sz+      !arr <- PM.unsafeFreezeByteArray marr1+      pure $! Right $! arr++-- | Receive a message. The message may be truncated if the size+-- of it is larger than the maximum allowed size.+tryReceive ::+     Connection -- ^ Connection+  -> Int -- ^ Maximum message size+  -> IO (Either (ReceiveException 'Uninterruptible) (Maybe ByteArray))+tryReceive (Connection !conn) !n = do+  !marr0 <- PM.newByteArray n+  Receive.receiveAttempt conn (MutableBytes marr0 0 n) >>= \case+    Left err -> pure (Left err)+    Right Nothing -> pure (Right Nothing)+    Right (Just sz) -> do+      marr1 <- PM.resizeMutableByteArray marr0 sz+      !arr <- PM.unsafeFreezeByteArray marr1+      pure $! Right $! Just $! arr++
+ src/Socket/SequencedPacket/Unix.hs view
@@ -0,0 +1,417 @@+{-# language BangPatterns #-}+{-# language DataKinds #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language MultiWayIf #-}+{-# language NamedFieldPuns #-}+{-# language UnboxedTuples #-}+{-# language ScopedTypeVariables #-}+{-# language TypeApplications #-}++-- | Unix-domain @seqpacket@ sockets.+module Socket.SequencedPacket.Unix+  ( Connection(..)+  , Listener(..)+  , BindException(..)+  , openPair+  , withPair+  , listen+  , unlisten+  , unlisten_+  , withListener+  , connect+  , tryConnect+  , withConnection+  , withAccepted+  , accept+  , disconnect+  , disconnect_+  ) where++import Control.Concurrent.STM (TVar)+import Control.Exception (mask,onException)+import Data.Coerce (coerce)+import Data.Primitive (ByteArray)+import Foreign.C.Error (Errno(..),eNOENT,ePROTOTYPE,eAGAIN,eWOULDBLOCK)+import Foreign.C.Error (eNFILE,eMFILE,eCONNREFUSED,eNOTCONN)+import Foreign.C.Error (eADDRINUSE,eACCES,eCONNABORTED,ePERM)+import Socket (Connectedness(..),Family(..),SocketException(..),BindException(..))+import Socket.Datagram (Socket(..))+import Socket.Stream (ConnectException(..),CloseException(..),AcceptException(..))+import Socket.Error (die)+import Socket.IPv4 (Message(..))+import Socket.Datagram.Common (close)+import Socket (Interruptibility(Uninterruptible))+import Socket.Datagram.Unix.Connected (UnixAddress(..))+import System.Posix.Types (Fd)++import qualified Control.Concurrent.STM as STM+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 Socket.EventManager as EM+import qualified Socket.Pair as Pair++newtype Connection = Connection Fd++newtype Listener = Listener Fd++-- | Unbracketed function for opening a connected socket pair. All warnings+-- that apply to 'open' apply to this function as well.+openPair :: IO (Either SocketException (Connection, Connection))+openPair = coerce+  @(IO (Either SocketException (Fd,Fd)))+  @(IO (Either SocketException (Connection,Connection)))+  (Pair.open S.sequencedPacket)++withPair ::+     (Connection -> Connection -> IO a)+     -- ^ Callback providing the connected datagram sockets+  -> IO (Either SocketException a)+withPair f = mask $ \restore -> openPair >>= \case+  Left err -> pure (Left err)+  Right (Connection fdA, Connection fdB) -> do+    a <- onException+      (restore (f (Connection fdA) (Connection fdB)))+      (S.uninterruptibleErrorlessClose fdA *> S.uninterruptibleErrorlessClose fdB)+    S.uninterruptibleClose fdA >>= \case+      Left err -> die ("Socket.Datagram.Unix.Connected.close: " ++ describeErrorCode err)+      Right _ -> S.uninterruptibleClose fdB >>= \case+        Left err -> die ("Socket.Datagram.Unix.Connected.close: " ++ describeErrorCode err)+        Right _ -> pure (Right a)++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"++-- | Open a socket that is used to listen for inbound connections.+withListener ::+     UnixAddress -- ^ Address to bind to (path or abstract namespace name)+  -> (Listener -> IO a) -- ^ Callback+  -> IO (Either (BindException 'Unix) a)+withListener endpoint f = mask $ \restore -> do+  listen endpoint >>= \case+    Left err -> pure (Left err)+    Right sck -> do+      a <- onException (restore (f sck)) (unlisten_ sck)+      unlisten sck+      pure (Right a)++-- | Open a socket that can be used to listen for inbound connections.+-- Requirements:+--+-- * This function may only be called in contexts where exceptions+--   are masked.+-- * The caller /must/ be sure to call 'unlistener' on the resulting+--   'Listener' exactly once to close underlying file descriptor.+-- * The 'Listener' cannot be used after being given as an argument+--   to 'unlistener'.+--+-- Noncompliant use of this function leads to undefined behavior. Prefer+-- 'withListener' unless you are writing an integration with a+-- resource-management library.+listen :: UnixAddress -> IO (Either (BindException 'Unix) Listener)+listen (UnixAddress remote) = do+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.sequencedPacket)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketListenException err+    Right fd -> do+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ remote+      S.uninterruptibleBind fd sockAddr >>= \case+        Left err -> do+          S.uninterruptibleErrorlessClose fd+          handleBindListenException err+        Right _ -> S.uninterruptibleListen fd 16 >>= \case+          -- We hardcode the listen backlog to 16. The author is unfamiliar+          -- with use cases where gains are realized from tuning this parameter.+          -- Open an issue if this causes problems for anyone.+          Left err -> do+            _ <- S.uninterruptibleClose fd+            handleBindListenException err+          Right _ -> do+            -- The getsockname is copied from code in Socket.Datagram.IPv4.Undestined.+            -- Consider factoring this out.+            let !mngr = EM.manager+            EM.register mngr fd+            pure (Right (Listener fd))++-- | Open a socket and connect to a peer. Requirements:+--+-- * This function may only be called in contexts where exceptions+--   are masked.+-- * The caller /must/ be sure to call 'disconnect' or 'disconnect_'+--   on the resulting 'Connection' exactly once to close underlying+--   file descriptor.+-- * The 'Connection' cannot be used after being given as an argument+--   to 'disconnect' or 'disconnect_'.+--+-- Noncompliant use of this function leads to undefined behavior. Prefer+-- 'withConnection' unless you are writing an integration with a+-- resource-management library.+connect ::+     UnixAddress+     -- ^ Peer address+  -> IO (Either (ConnectException 'Unix 'Uninterruptible) Connection)+connect (UnixAddress !remote) = do+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.sequencedPacket)+    S.defaultProtocol+  case e1 of+    Left err -> handleConnectException err+    Right fd -> do+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ remote+      let !mngr = EM.manager+      -- TODO: I believe it is sound to make both the write and+      -- read channels start off as not ready. After all, the+      -- socket is brand new and is not connected to a peer.+      -- Consequently, there's no way we could miss events.+      EM.register mngr fd+      -- This is currently wrong. Redo this later.+      S.uninterruptibleConnect fd sockAddr >>= \case+        Left err2 -> do+          S.uninterruptibleErrorlessClose fd+          handleConnectException err2+        Right _ -> pure (Right (Connection fd))++-- | Variant of 'connect' that does not block. Returns 'Nothing' if+-- the connection cannot be established immidiately.+tryConnect ::+     UnixAddress+     -- ^ Peer address+  -> IO (Either (ConnectException 'Unix 'Uninterruptible) (Maybe Connection))+tryConnect (UnixAddress !remote) = do+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.sequencedPacket)+    S.defaultProtocol+  case e1 of+    Left err -> handleConnectException err+    Right fd -> do+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ remote+      let !mngr = EM.manager+      -- TODO: I believe it is sound to make both the write and+      -- read channels start off as not ready. After all, the+      -- socket is brand new and is not connected to a peer.+      -- Consequently, there's no way we could miss events.+      EM.register mngr fd+      -- This is currently wrong. Redo this later.+      S.uninterruptibleConnect fd sockAddr >>= \case+        Left err2 -> do+          S.uninterruptibleErrorlessClose fd+          if | err2 == eAGAIN -> pure (Right Nothing)+             | err2 == eWOULDBLOCK -> pure (Right Nothing)+             | otherwise -> handleConnectException err2+        Right _ -> pure (Right (Just (Connection fd)))++-- These are the exceptions that can happen as a result+-- of calling @socket@ with the intent of using the socket+-- to listen for inbound connections.+handleSocketListenException :: Errno -> IO (Either (BindException 'Unix) a)+handleSocketListenException e+  | e == eMFILE = pure (Left BindFileDescriptorLimit)+  | e == eNFILE = pure (Left BindFileDescriptorLimit)+  | otherwise = die ("Socket.SequencedPacket.Unix.socket: " ++ describeErrorCode e)++-- These are the exceptions that can happen as a result+-- of calling @bind@ with the intent of using the socket+-- to listen for inbound connections. This is also used+-- to clean up the error codes of @listen@. The two can+-- report some of the same error codes, and those happen+-- to be the error codes we are interested in.+--+-- NB: EACCES only happens on @bind@, not on @listen@.+handleBindListenException :: Errno -> IO (Either (BindException 'Unix) a)+handleBindListenException !e+  | e == eACCES = pure (Left BindPermissionDenied)+  | e == eADDRINUSE = pure (Left BindAddressInUse)+  | otherwise = die ("Socket.SequencedPacket.Unix.bindListen: " ++ describeErrorCode e)++handleConnectException :: Errno -> IO (Either (ConnectException 'Unix i)  a)+handleConnectException e+  | e == eNOENT = pure (Left ConnectNoEntry)+  | e == ePROTOTYPE = pure (Left ConnectProtocolType)+  | e == eCONNREFUSED = pure (Left ConnectRefused)+  | otherwise = die+      ("Socket.Datagram.Unix.Connected.connect: " ++ describeErrorCode e)++-- | Close a listener. This does not check to see whether or not+-- the operating system successfully closed the socket. It never+-- throws exceptions of any kind. This should only be preferred+-- to 'unlistener' in exception-cleanup contexts where there is+-- already an exception that will be rethrown. See the implementation+-- of 'withListener' for an example of appropriate use of both+-- 'unlistener' and 'unlistener_'.+unlisten_ :: Listener -> IO ()+unlisten_ (Listener fd) = S.uninterruptibleErrorlessClose fd++-- | Close a listener. This throws an unrecoverable exception if+--   the socket cannot be closed.+unlisten :: Listener -> IO ()+unlisten (Listener fd) = S.uninterruptibleClose fd >>= \case+  Left err -> die ("Socket.SequencedPacket.Unix.unlisten: " ++ describeErrorCode err)+  Right _ -> pure ()++-- | Establish a connection to a server.+withConnection ::+     UnixAddress+     -- ^ Peer address+  -> (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 'Unix 'Uninterruptible) b)+withConnection !remote g f = mask $ \restore -> do+  connect 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)+    +-- | 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+-- resets need to be sent.+disconnect :: Connection -> IO (Either CloseException ())+disconnect (Connection fd) = gracefulCloseA fd++gracefulCloseA :: Fd -> IO (Either CloseException ())+gracefulCloseA fd = do+  let !mngr = EM.manager+  !tv <- EM.reader mngr fd+  token0 <- STM.readTVarIO tv+  S.uninterruptibleShutdown fd S.write >>= \case+    -- On Linux (not sure about others), calling shutdown+    -- on the write channel fails with with ENOTCONN if the+    -- write channel is already closed. It is common for this to+    -- happen (e.g. if the peer calls @close@ before the local+    -- process runs gracefulClose, the local operating system+    -- will have already closed the write channel). However,+    -- it does not pose a problem. We just proceed as we would+    -- have since either way we become certain that the write channel+    -- is closed.+    Left err -> if err == eNOTCONN+      then gracefulCloseB tv token0 fd+      else do+        S.uninterruptibleErrorlessClose fd+        die ("Socket.SequencedPacket.Unix.gracefulCloseA[shutdown]: " ++ describeErrorCode err)+    Right _ -> gracefulCloseB tv token0 fd++-- The second part of the shutdown function must call itself recursively+-- since we may receive false read-ready notifications at any time.+gracefulCloseB :: TVar EM.Token -> EM.Token -> Fd -> IO (Either CloseException ())+gracefulCloseB !tv !token0 !fd = do+  !buf <- PM.newByteArray 1+  S.uninterruptibleReceiveMutableByteArray fd buf 0 1 mempty >>= \case+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+      then do+        token1 <- EM.persistentUnreadyAndWait token0 tv+        gracefulCloseB tv token1 fd+      else do+        _ <- S.uninterruptibleClose fd+        -- We treat all @recv@ errors except for the nonblocking+        -- notices as unrecoverable.+        die ("Socket.SequencedPacket.Unix.gracefulCloseB[recv]: " ++ describeErrorCode err1)+    Right sz -> if sz == 0+      then S.uninterruptibleClose fd >>= \case+        Left err -> +          die ("Socket.SequencedPacket.Unix.gracefulCloseB[close]: " ++ describeErrorCode err)+        Right _ -> pure (Right ())+      else do+        _ <- S.uninterruptibleClose fd+        pure (Left ClosePeerContinuedSending)++-- | Close a connection. This does not check to see whether or not+-- the connection was brought down gracefully. It just calls @close@.+-- It never throws exceptions of any kind (even if @close@ fails).+-- This should only be preferred to 'disconnect' in exception-cleanup+-- contexts where there is already an exception that will be rethrown.+-- See the implementation of 'withConnection' for an example of+-- appropriate use of both 'disconnect' and 'disconnect_'.+disconnect_ :: Connection -> IO ()+disconnect_ (Connection fd) = S.uninterruptibleErrorlessClose fd++-- | Accept a connection on the listener and run the supplied callback+-- on it. This closes the connection when the callback finishes or if+-- an exception is thrown. Since this function blocks the thread until+-- the callback finishes, it is only suitable for stream socket clients+-- that handle one connection at a time. The variant 'forkAcceptedUnmasked'+-- is preferrable for servers that need to handle connections concurrently+-- (most use cases).+withAccepted ::+     Listener+  -> (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 (AcceptException 'Uninterruptible) b)+withAccepted !lstn consumeException cb = do+  r <- mask $ \restore -> do+    accept lstn >>= \case+      Left e -> pure (Left e)+      Right conn -> do+        a <- onException (restore (cb conn)) (disconnect_ conn)+        e <- disconnect conn+        pure (Right (e,a))+  -- Notice that consumeException gets run in an unmasked context.+  case r of+    Left e -> pure (Left e)+    Right (e,a) -> fmap Right (consumeException e a)++-- | Listen for an inbound connection.+accept :: Listener -> IO (Either (AcceptException 'Uninterruptible) Connection)+accept (Listener !fd) = do+  -- Although this function must be called in a context where+  -- exceptions are masked, recall that EM.wait uses an STM+  -- action that might retry, meaning that this first part is+  -- still interruptible. This is a good thing in the case of+  -- this function.+  let !mngr = EM.manager+  -- The listener should already be registered, so we can just+  -- ask for the reader directly.+  !tv <- EM.reader mngr fd+  let go !oldToken = do+        waitlessAccept fd >>= \case+          Left merr -> case merr of+            Nothing -> EM.unreadyAndWait oldToken tv >>= go+            Just err -> pure (Left err)+          Right r@(Connection conn) -> do+            EM.register mngr conn+            pure (Right r)+  go =<< STM.readTVarIO tv++waitlessAccept :: Fd -> IO (Either (Maybe (AcceptException i)) Connection)+waitlessAccept !lstn = do+  -- TODO: add a variant of accept4 in posix-api that discards the peer address.+  L.uninterruptibleAccept4 lstn 128 (L.closeOnExec <> L.nonblocking) >>= \case+    Left err -> handleAcceptException err+    Right (_,_,acpt) -> pure (Right (Connection acpt))++-- These are the exceptions that can happen as a result+-- of calling @accept@.+-- TODO: There is no way a UNIX-domain connection could be firewalled.+-- Is EPERM even possible in this context?+handleAcceptException :: Errno -> IO (Either (Maybe (AcceptException i)) a)+handleAcceptException e+  | e == eAGAIN = pure (Left Nothing)+  | e == eWOULDBLOCK = pure (Left Nothing)+  | e == eCONNABORTED = pure (Left (Just AcceptConnectionAborted))+  | e == eMFILE = pure (Left (Just AcceptFileDescriptorLimit))+  | e == eNFILE = pure (Left (Just AcceptFileDescriptorLimit))+  | e == ePERM = pure (Left (Just AcceptFirewalled))+  | otherwise = die ("Socket.Stream.IPv4.accept: " ++ describeErrorCode e)
src/Socket/Stream/IPv4.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE UnboxedTuples #-} {-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-}@@ -8,15 +10,18 @@ {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}  module Socket.Stream.IPv4   ( -- * Types-    Listener+    Listener(..)   , Connection(..)   , Peer(..)     -- * Bracketed     -- $brackeded   , withListener+  , withListenerReuse   , withAccepted   , withConnection   , forkAccepted@@ -29,6 +34,7 @@   , SocketException(..)   , AcceptException(..)   , CloseException(..)+  , SystemdException(..)     -- * Type Arguments   , Interruptibility(..)   , Family(..)@@ -36,14 +42,20 @@     -- * Unbracketed     -- $unbracketed   , listen+  , listenReuse   , unlisten   , unlisten_   , connect-  -- , interruptibleConnect+  , interruptibleConnect+  , connectOnDevice+  , interruptibleConnectOnDevice+  , systemdListener   , disconnect   , disconnect_   , accept   , interruptibleAccept+    -- * Shutdown+  , shutdown   ) where  import Control.Concurrent (ThreadId)@@ -51,12 +63,14 @@ import Control.Exception (mask, mask_, onException, throwIO) import Control.Monad.STM (atomically) import Control.Concurrent.STM (TVar,modifyTVar')+import Data.Coerce (coerce) import Data.Word (Word16) import Foreign.C.Error (Errno(..), eAGAIN, eINPROGRESS, eWOULDBLOCK, eNOTCONN) import Foreign.C.Error (eADDRINUSE,eHOSTUNREACH) import Foreign.C.Error (eNFILE,eMFILE,eACCES,ePERM,eCONNABORTED) import Foreign.C.Error (eTIMEDOUT,eADDRNOTAVAIL,eNETUNREACH,eCONNREFUSED) import Foreign.C.Types (CInt)+import GHC.Exts (Int#) import Net.Types (IPv4(..)) import Socket (Interruptibility(..)) import Socket (SocketUnrecoverableException(..),Family(Internet),Version(V4))@@ -67,7 +81,8 @@ import Socket.Stream (ConnectException(..),SocketException(..),AcceptException(..)) import Socket.Stream (SendException(..),ReceiveException(..),CloseException(..)) import Socket.Stream (Connection(..))-import System.Posix.Types(Fd)+import Socket.Systemd (SystemdException(..),systemdListenerInternal)+import System.Posix.Types (Fd(Fd))  import qualified Control.Concurrent.STM as STM import qualified Data.Primitive as PM@@ -94,15 +109,31 @@ -- 'withListener' unless you are writing an integration with a -- resource-management library. listen :: Peer -> IO (Either SocketException (Listener, Word16))-listen endpoint@Peer{port = specifiedPort} = do+listen !peer = listenInternal 0# peer++-- | Variant of 'listen' that sets @SO_REUSEADDR@ on the socket before+-- binding.+listenReuse :: Peer -> IO (Either SocketException (Listener, Word16))+listenReuse !peer = listenInternal 1# peer++listenInternal :: Int# -> Peer -> IO (Either SocketException (Listener, Word16))+listenInternal reuse endpoint@Peer{port = specifiedPort} = do   debug ("listen: opening listen " ++ describeEndpoint endpoint)-  e1 <- S.uninterruptibleSocket S.internet+  e1 <- S.uninterruptibleSocket S.Internet     (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream)     S.defaultProtocol   debug ("listen: opened listen " ++ describeEndpoint endpoint)   case e1 of     Left err -> handleSocketListenException SCK.functionWithListener err     Right fd -> do+      case reuse of+        0# -> pure ()+        _ -> do+          r <- S.uninterruptibleSetSocketOptionInt fd S.levelSocket S.reuseAddress 1+          case r of+            Left _ -> die "listenInternal: setsockopt SO_REUSEADDR failed"+            Right (_ :: ()) -> pure ()+      -- TODO: shave off an allocation by building the sockaddr in C.       e2 <- S.uninterruptibleBind fd         (S.encodeSocketAddressInternet (endpointToSocketAddressInternet endpoint))       debug ("listen: requested binding for listen " ++ describeEndpoint endpoint)@@ -176,7 +207,7 @@      Peer   -> (Listener -> Word16 -> IO a)   -> IO (Either SocketException a)-withListener endpoint f = mask $ \restore -> do+withListener !endpoint f = mask $ \restore -> do   listen endpoint >>= \case     Left err -> pure (Left err)     Right (sck, actualPort) -> do@@ -186,6 +217,21 @@       unlisten sck       pure (Right a) +-- | Variant of 'withListener' that sets @SO_REUSEADDR@ before binding.+withListenerReuse ::+     Peer+  -> (Listener -> Word16 -> IO a)+  -> IO (Either SocketException a)+withListenerReuse !endpoint f = mask $ \restore -> do+  listenReuse endpoint >>= \case+    Left err -> pure (Left err)+    Right (sck, actualPort) -> do+      a <- onException+        (restore (f sck actualPort))+        (unlisten_ sck)+      unlisten sck+      pure (Right a)+ -- | Listen for an inbound connection. accept :: Listener -> IO (Either (AcceptException 'Uninterruptible) (Connection,Peer)) accept (Listener !fd) = do@@ -247,13 +293,16 @@   -- TODO: pull these out of the loop   let !mngr = EM.manager   tv <- EM.reader mngr fd-  token <- EM.interruptibleWait abandon tv+  token <- EM.interruptibleWaitCounting counter abandon tv   if EM.isInterrupt token     then pure (Left AcceptInterrupted)     else waitlessAccept fd >>= \case       Left merr -> case merr of         Nothing -> do           EM.unready token tv+          -- Decrement the connection counter if the notification+          -- from epoll was a false alarm.+          atomically (modifyTVar' counter (subtract 1))           interruptibleAcceptCounting counter abandon (Listener fd)         Just err -> pure (Left err)       Right r@(Connection conn,_) -> do@@ -284,11 +333,20 @@           SCK.functionWithAccepted           [SCK.cgetsockname,SCK.socketAddressSize] +-- | Close the write channel. Any attempt to write to the connection after+-- calling this will fail.+shutdown :: Connection -> IO ()+shutdown (Connection fd) = do+  S.uninterruptibleShutdown fd S.write >>= \case+    Left err -> if err == eNOTCONN+      then pure ()+      else do+        S.uninterruptibleErrorlessClose fd+        die ("Socket.Stream.shutdown: " ++ describeErrorCode err)+    Right _ -> pure ()+ gracefulCloseA :: Fd -> IO (Either CloseException ()) gracefulCloseA fd = do-  let !mngr = EM.manager-  !tv <- EM.reader mngr fd-  token0 <- STM.readTVarIO tv   S.uninterruptibleShutdown fd S.write >>= \case     -- On Linux (not sure about others), calling shutdown     -- on the write channel fails with with ENOTCONN if the@@ -300,30 +358,38 @@     -- have since either way we become certain that the write channel     -- is closed.     Left err -> if err == eNOTCONN-      then gracefulCloseB tv token0 fd+      then gracefulCloseB fd       else do-        _ <- S.uninterruptibleClose fd-        -- TODO: What about ENOTCONN? Can this happen if the remote-        -- side has already closed the connection?+        S.uninterruptibleErrorlessClose fd         throwIO $ SocketUnrecoverableException           moduleSocketStreamIPv4           SCK.functionGracefulClose           [SCK.cshutdown,describeErrorCode err]-    Right _ -> gracefulCloseB tv token0 fd+    Right _ -> gracefulCloseB fd --- The second part of the shutdown function must call itself recursively--- since we may receive false read-ready notifications at any time.-gracefulCloseB :: TVar EM.Token -> EM.Token -> Fd -> IO (Either CloseException ())-gracefulCloseB !tv !token0 !fd = do+-- In the commit after 30c0037b99517b4e665aec33ac3a479fc892cb98,+-- gracefulCloseB was changed so that it does not wait for the peer to+-- shut down its side of the connection. Although this behavior was useful,+-- it made it possible for an unresponsive peer to hold open the connection.+-- The new behavior is to make a single nonblocking attempt to read from+-- the socket. If bytes are available, then we return a CloseException.+-- Otherwise, we return Right. This is less deterministic than the+-- previous approach (network latency affects the result), but the+-- exceptions returned here are really only ever used for logging+-- purposes anyway.+gracefulCloseB :: Fd -> IO (Either CloseException ())+gracefulCloseB !fd = do   !buf <- PM.newByteArray 1   -- We do not actually want to remove the bytes from the-  -- receive buffer, so we use MSG_PEEK. We are be certain+  -- receive buffer, so we use MSG_PEEK. We are certain   -- to send a reset when a CloseException is reported.+  -- Retrospective: Why is MSG_PEEK important? Who cares if the+  -- bytes get eaten? The receive buffer is about to get axed anyway.   S.uninterruptibleReceiveMutableByteArray fd buf 0 1 S.peek >>= \case-    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN || err1 == eNOTCONN       then do-        token1 <- EM.persistentUnreadyAndWait token0 tv-        gracefulCloseB tv token1 fd+        _ <- S.uninterruptibleClose fd+        pure (Right ())       else do         _ <- S.uninterruptibleClose fd         -- We treat all @recv@ errors except for the nonblocking@@ -470,17 +536,14 @@   -> Listener      -- ^ Connection listener   -> (Either CloseException () -> a -> IO ())-     -- ^ Callback to handle an ungraceful close. +     -- ^ Callback to handle an ungraceful close. This must not+     --   throw an exception.   -> (Connection -> Peer -> IO a)      -- ^ Callback to consume connection. Must not return the connection.   -> IO (Either (AcceptException 'Interruptible) ThreadId) interruptibleForkAcceptedUnmasked !counter !abandon !lstn consumeException cb =   mask_ $ interruptibleAcceptCounting counter abandon lstn >>= \case-    Left e -> do-      case e of-        AcceptInterrupted -> pure ()-        _ -> atomically (modifyTVar' counter (subtract 1))-      pure (Left e)+    Left e -> pure (Left e)     Right (conn, endpoint) -> fmap Right $ forkIOWithUnmask $ \unmask -> do       a <- onException         (unmask (cb conn endpoint))@@ -490,6 +553,49 @@       atomically (modifyTVar' counter (subtract 1))       pure r +-- | Variant of 'connect' that allows specifying the name of the local+-- interface that should be used for the connection. Uses @SO_BINDTODEVICE@.+connectOnDevice ::+     Peer -- ^ Remote endpoint+  -> PM.ByteArray -- ^ Local network interface name+  -> IO (Either (ConnectException ('Internet 'V4) 'Uninterruptible) Connection)+connectOnDevice !remote !intf = do+  beforeEstablishment remote >>= \case+    Left err -> pure (Left err)+    Right (fd,sockAddr) -> do+      r <- S.uninterruptibleSetSocketOptionByteArray+        fd S.levelSocket S.bindToDevice+        intf (fromIntegral @Int @CInt (PM.sizeofByteArray intf))+      -- We throw a bogus error here. TODO: fix this.+      case r of+        Left _ -> do+          S.uninterruptibleErrorlessClose fd+          pure (Left ConnectEphemeralPortsExhausted)+        Right (_ :: ()) -> prepareConnection fd sockAddr++-- | Variant of 'interruptibleConnect' that allows specifying the name of the local+-- interface that should be used for the connection. Uses @SO_BINDTODEVICE@.+interruptibleConnectOnDevice ::+     TVar Bool+     -- ^ Interrupted. If this becomes 'True', give up and return+     --   @'Left' 'AcceptInterrupted'@.+  -> Peer -- ^ Remote endpoint+  -> PM.ByteArray -- ^ Local network interface name+  -> IO (Either (ConnectException ('Internet 'V4) 'Interruptible) Connection)+interruptibleConnectOnDevice !abandon !remote !intf = do+  beforeEstablishment remote >>= \case+    Left err -> pure (Left err)+    Right (fd,sockAddr) -> do+      r <- S.uninterruptibleSetSocketOptionByteArray+        fd S.levelSocket S.bindToDevice+        intf (fromIntegral @Int @CInt (PM.sizeofByteArray intf))+      -- We throw a bogus error here. TODO: fix this.+      case r of+        Left _ -> do+          S.uninterruptibleErrorlessClose fd+          pure (Left ConnectEphemeralPortsExhausted)+        Right (_ :: ()) -> interruptiblePrepareConnection abandon fd sockAddr+ -- | Open a socket and connect to a peer. Requirements: -- -- * This function may only be called in contexts where exceptions@@ -510,60 +616,88 @@ connect !remote = do   beforeEstablishment remote >>= \case     Left err -> pure (Left err)-    Right (fd,sockAddr) -> do-      let !mngr = EM.manager-      -- TODO: I believe it is sound to make both the write and-      -- read channels start off as not ready. After all, the-      -- socket is brand new and is not connected to a peer.-      -- Consequently, there's no way we could miss events.-      EM.register mngr fd-      tv <- EM.writer mngr fd-      debug ("connect: about to connect, fd=" ++ show fd)-      token0 <- STM.readTVarIO tv-      S.uninterruptibleConnect fd sockAddr >>= \case-        Left err2 -> if err2 == eINPROGRESS-          then do-            debug ("connect: EINPROGRESS, fd=" ++ show fd)-            -- When we receive EINPROGRESS, we have high confidence-            -- the that the socket is not yet ready (keeping in mind-            -- that it could have somehow become ready right after-            -- C's connect returned). Immidiately diving into-            -- afterEstablishment would result in a syscall to-            -- getsockopt. But what a waste that would be since it-            -- would almost certainly return EAGAIN. So, instead,-            -- we unready the write channel (taking the usual-            -- precautions) and then await readiness.-            token1 <- EM.unreadyAndWait token0 tv-            afterEstablishment tv token1 fd-          else do-            debug ("connect: failed, fd=" ++ show fd)+    Right (fd,sockAddr) -> prepareConnection fd sockAddr++prepareConnection ::+     Fd+  -> S.SocketAddress+  -> IO (Either (ConnectException ('Internet 'V4) 'Uninterruptible) Connection)+prepareConnection !fd !sockAddr = do+  let !mngr = EM.manager+  -- TODO: I believe it is sound to make both the write and+  -- read channels start off as not ready. After all, the+  -- socket is brand new and is not connected to a peer.+  -- Consequently, there's no way we could miss events.+  EM.register mngr fd+  tv <- EM.writer mngr fd+  debug ("connect: about to connect, fd=" ++ show fd)+  token0 <- STM.readTVarIO tv+  S.uninterruptibleConnect fd sockAddr >>= \case+    Left err2 -> if err2 == eINPROGRESS+      then do+        debug ("connect: EINPROGRESS, fd=" ++ show fd)+        -- When we receive EINPROGRESS, we have high confidence+        -- the that the socket is not yet ready (keeping in mind+        -- that it could have somehow become ready right after+        -- C's connect returned). Immidiately diving into+        -- afterEstablishment would result in a syscall to+        -- getsockopt. But what a waste that would be since it+        -- would almost certainly return EAGAIN. So, instead,+        -- we unready the write channel (taking the usual+        -- precautions) and then await readiness.+        token1 <- EM.unreadyAndWait token0 tv+        afterEstablishment tv token1 fd+      else do+        debug ("connect: failed, fd=" ++ show fd)+        S.uninterruptibleErrorlessClose fd+        handleConnectException SCK.functionWithConnection err2+    Right _ -> do+      debug ("connect: succeeded immidiately, fd=" ++ show fd)+      afterEstablishment tv token0 fd++-- See the notes in prepareConnection+interruptiblePrepareConnection ::+     TVar Bool+  -> Fd+  -> S.SocketAddress+  -> IO (Either (ConnectException ('Internet 'V4) 'Interruptible) Connection)+interruptiblePrepareConnection !abandon !fd !sockAddr = do+  let !mngr = EM.manager+  EM.register mngr fd+  tv <- EM.writer mngr fd+  debug ("connect: about to connect, fd=" ++ show fd)+  token0 <- STM.readTVarIO tv+  S.uninterruptibleConnect fd sockAddr >>= \case+    Left err2 -> if err2 == eINPROGRESS+      then do+        debug ("connect: EINPROGRESS, fd=" ++ show fd)+        EM.unready token0 tv+        token1 <- EM.interruptibleWait abandon tv+        case EM.isInterrupt token1 of+          True -> do             S.uninterruptibleErrorlessClose fd-            handleConnectException SCK.functionWithConnection err2-        Right _ -> do-          debug ("connect: succeeded immidiately, fd=" ++ show fd)-          afterEstablishment tv token0 fd+            pure (Left ConnectInterrupted)+          False -> interruptibleAfterEstablishment abandon tv token1 fd+      else do+        debug ("connect: failed, fd=" ++ show fd)+        S.uninterruptibleErrorlessClose fd+        handleConnectException SCK.functionWithConnection err2+    Right _ -> do+      debug ("connect: succeeded immidiately, fd=" ++ show fd)+      interruptibleAfterEstablishment abandon tv token0 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'@.---   -> Peer---      -- ^ 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+interruptibleConnect ::+     TVar Bool+     -- ^ Interrupted. If this becomes 'True', give up and return+     --   @'Left' 'AcceptInterrupted'@.+  -> Peer+     -- ^ Remote endpoint+  -> IO (Either (ConnectException ('Internet 'V4) 'Interruptible) Connection)+interruptibleConnect !abandon !remote = do+  beforeEstablishment remote >>= \case+    Left err -> pure (Left err)+    Right (fd,sockAddr) -> interruptiblePrepareConnection abandon fd sockAddr  -- Internal function called by both connect and interruptibleConnect -- before the connection is established. Creates the socket and prepares@@ -572,7 +706,7 @@ {-# INLINE beforeEstablishment #-} beforeEstablishment !remote = do   debug ("beforeEstablishment: opening connection " ++ show remote)-  e1 <- S.uninterruptibleSocket S.internet+  e1 <- S.uninterruptibleSocket S.Internet     (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream)     S.defaultProtocol   debug ("beforeEstablishment: opened connection " ++ show remote)@@ -624,6 +758,48 @@           functionWithListener           [SCK.cgetsockopt,connectErrorOptionValueSize] +interruptibleAfterEstablishment ::+     TVar Bool+  -> TVar EM.Token -- token tvar +  -> EM.Token -- old token+  -> Fd+  -> IO (Either (ConnectException ('Internet 'V4) 'Interruptible) Connection)+interruptibleAfterEstablishment !abandon !tv !oldToken !fd = do+  debug ("interruptibleAfterEstablishment: finished waiting, fd=" ++ show fd)+  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 -> do+               debug ("interruptibleAfterEstablishment: connection established, fd=" ++ show fd)+               pure (Right (Connection fd))+           | Errno err == eAGAIN || Errno err == eWOULDBLOCK -> do+               debug ("interruptibleAfterEstablishment: not ready yet, unreadying token and waiting, fd=" ++ show fd)+               EM.unready oldToken tv+               newToken <- EM.interruptibleWait abandon tv+               case EM.isInterrupt newToken of+                 True -> do+                   S.uninterruptibleErrorlessClose fd+                   pure (Left ConnectInterrupted)+                 False -> interruptibleAfterEstablishment abandon tv newToken fd+           | otherwise -> 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@@ -759,8 +935,7 @@       ("Socket.Stream.IPv4.bindListen: " ++ describeErrorCode e)  -- These are the exceptions that can happen as a result--- of calling @socket@ with the intent of using the socket--- to open a connection (not listen for inbound connections).+-- of calling @accept@. handleAcceptException :: Errno -> IO (Either (Maybe (AcceptException i)) a) handleAcceptException e   | e == eAGAIN = pure (Left Nothing)@@ -773,6 +948,20 @@  connectErrorOptionValueSize :: String connectErrorOptionValueSize = "incorrectly sized value of SO_ERROR option"++-- | Retrieve a listener that systemd has passed to the process. This+-- may only be called once. Do not call 'unlisten' or 'unlisten_' on this+-- listener before exiting the application. There is no bracketed variant+-- of this function because the listener is expected to remain open for+-- the remainder of the application.+--+-- There are several reasons this function may return @Left@:+--+-- * @sd_listen_fds@ returned more than one file descriptor+-- * @sd_is_socket@ found that the file descriptor was not a socket or+--   that it was a socket that was not in listening mode.+systemdListener :: IO (Either SystemdException Listener)+systemdListener = coerce (systemdListenerInternal S.Internet)  {- $bracketed
src/Socket/Stream/Interruptible/Bytes.hs view
@@ -6,6 +6,7 @@   ( send   , receiveExactly   , receiveOnce+  , receiveOncePrepended   , receiveBetween   ) where @@ -16,6 +17,7 @@ import Socket (Interruptibility(Interruptible))  import qualified Data.Primitive as PM+import qualified Data.Bytes as Bytes import qualified Socket.Stream.Interruptible.Bytes.Send as Send import qualified Socket.Stream.Interruptible.MutableBytes.Receive as Receive @@ -64,6 +66,28 @@     Left err -> pure (Left err)     Right sz -> do       marr1 <- PM.resizeMutableByteArray marr0 sz+      !arr <- PM.unsafeFreezeByteArray marr1+      pure $! Right $! arr++-- | Variant of 'receiveOnce' that returns prepends the received bytes+-- with a prefix. This can be used to attach feed leftovers back in when+-- processing line-oriented protocols.+receiveOncePrepended ::+     TVar Bool +     -- ^ Interrupt. On 'True', give up and return @'Left' 'ReceiveInterrupted'@.+  -> Connection -- ^ Connection+  -> Bytes -- ^ Prefix to prepend to the received bytes+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either (ReceiveException 'Interruptible) ByteArray)+{-# inline receiveOncePrepended #-}+receiveOncePrepended !tv !conn !prefix !n = do+  let !ix0 = Bytes.length prefix+  !marr0 <- PM.newByteArray (n + ix0)+  Bytes.unsafeCopy marr0 0 prefix+  Receive.receiveOnce tv conn (MutableBytes marr0 ix0 n) >>= \case+    Left err -> pure (Left err)+    Right sz -> do+      marr1 <- PM.resizeMutableByteArray marr0 (sz + ix0)       !arr <- PM.unsafeFreezeByteArray marr1       pure $! Right $! arr
src/Socket/Stream/Uninterruptible/Bytes.hs view
@@ -12,8 +12,10 @@ -- be zero and the length would be the length of the 'ByteArray' -- payload. module Socket.Stream.Uninterruptible.Bytes-  ( send+  ( -- * Send+    send   , sendMany+    -- * Receive   , receiveExactly   , receiveOnce   , receiveBetween
src/Socket/Stream/Uninterruptible/MutableBytes.hs view
@@ -3,17 +3,23 @@ {-# language MagicHash #-}  module Socket.Stream.Uninterruptible.MutableBytes-  ( send+  ( -- * Send+    send+  , sendMany+    -- * Receive   , receiveExactly   , receiveOnce   , receiveBetween   ) where  import Data.Bytes.Types (MutableBytes)+import Data.Primitive (MutableByteArray)+import Data.Primitive.Unlifted.Array (UnliftedArray(..)) import GHC.Exts (RealWorld,proxy#) import Socket.Stream (Connection,ReceiveException,SendException) import Socket (Interruptibility(Uninterruptible)) +import qualified Socket.Stream.Uninterruptible.Bytes as SSUB import qualified Socket.Stream.Uninterruptible.MutableBytes.Send as Send import qualified Socket.Stream.Uninterruptible.MutableBytes.Receive as Receive @@ -25,6 +31,22 @@   -> IO (Either (SendException 'Uninterruptible) ()) {-# inline send #-} send = Send.send proxy#++-- | 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 (MutableByteArray RealWorld) -- ^ Byte arrays+  -> IO (Either (SendException 'Uninterruptible) ())+{-# inline sendMany #-}+sendMany conn (UnliftedArray bufs) =+  -- Effectively, what we are doing here is unsafely coercing the+  -- buffer from mutable to immutable. This ends up being sound+  -- in this context because SSUB.sendMany does not preserve any+  -- referenced to its argument byte arrays.+  SSUB.sendMany conn (UnliftedArray bufs)  -- | Receive a number of bytes exactly equal to the length of the --   buffer slice. If needed, this may call @recv@ repeatedly until
+ src/Socket/Stream/Unix.hs view
@@ -0,0 +1,361 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}++module Socket.Stream.Unix+  ( -- * Types+    Listener(..)+  , Connection(..)+  , UnixAddress(..)+    -- * Bracketed+  , withListener+  , withAccepted+    -- * Exceptions+  , SendException(..)+  , ReceiveException(..)+  , ConnectException(..)+  , SocketException(..)+  , AcceptException(..)+  , CloseException(..)+  , SystemdException(..)+    -- * Bracketed+  , interruptibleForkAcceptedUnmasked+    -- * Unbracketed+    -- $unbracketed+  , listen+  , unlisten+  , unlisten_+  , disconnect+  , disconnect_+  , accept+  , systemdListener+  ) where++import Control.Concurrent (ThreadId,forkIOWithUnmask)+import Control.Concurrent.STM (TVar,modifyTVar',atomically)+import Control.Exception (mask, mask_, onException)+import Data.Coerce (coerce)+import Foreign.C.Error (Errno(..), eAGAIN, eWOULDBLOCK, eNOTCONN)+import Foreign.C.Error (eADDRINUSE)+import Foreign.C.Error (eNFILE,eMFILE,eACCES,ePERM,eCONNABORTED)+import Socket.Datagram.Unix.Connected (UnixAddress(..))+import Socket.Error (die)+import Socket.Stream (SocketException(..),AcceptException(..),ConnectException(..))+import Socket.Stream (Connection(..))+import Socket.Stream (SendException(..),ReceiveException(..),CloseException(..))+import System.Posix.Types (Fd(Fd))+import Socket (Interruptibility(..))+import Socket.Systemd (SystemdException(..),systemdListenerInternal)+import qualified Control.Concurrent.STM as STM+import qualified Foreign.C.Error.Describe as D+import qualified Socket.EventManager as EM+import qualified Posix.Socket as S+import qualified Linux.Socket as L+import qualified Data.Primitive as PM++-- | A socket that listens for incomming connections.+newtype Listener = Listener Fd++-- | Open a socket that can be used to listen for inbound connections.+-- Requirements:+--+-- * This function may only be called in contexts where exceptions+--   are masked.+-- * The caller /must/ be sure to call 'unlistener' on the resulting+--   'Listener' exactly once to close underlying file descriptor.+-- * The 'Listener' cannot be used after being given as an argument+--   to 'unlistener'.+--+-- Noncompliant use of this function leads to undefined behavior. Prefer+-- 'withListener' unless you are writing an integration with a+-- resource-management library.+listen :: UnixAddress -> IO (Either SocketException Listener)+listen (UnixAddress path) = do+  e1 <- S.uninterruptibleSocket S.Unix+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream)+    S.defaultProtocol+  case e1 of+    Left err -> handleSocketListenException err+    Right fd -> do+      let sockAddr = id+            $ S.encodeSocketAddressUnix+            $ S.SocketAddressUnix+            $ path+      e2 <- S.uninterruptibleBind fd sockAddr+      case e2 of+        Left err -> do+          _ <- S.uninterruptibleClose fd+          handleBindListenException err+        Right _ -> S.uninterruptibleListen fd 16 >>= \case+          -- We hardcode the listen backlog to 16. The author is unfamiliar+          -- with use cases where gains are realized from tuning this parameter.+          -- Open an issue if this causes problems for anyone.+          Left err -> do+            _ <- S.uninterruptibleClose fd+            handleBindListenException err+          Right _ -> do+            let !mngr = EM.manager+            EM.register mngr fd+            pure (Right (Listener fd))++-- These are the exceptions that can happen as a result+-- of calling @bind@ with the intent of using the socket+-- to listen for inbound connections. This is also used+-- to clean up the error codes of @listen@. The two can+-- report some of the same error codes, and those happen+-- to be the error codes we are interested in.+--+-- NB: EACCES only happens on @bind@, not on @listen@.+handleBindListenException :: Errno -> IO (Either SocketException a)+handleBindListenException !e+  | e == eACCES = pure (Left SocketPermissionDenied)+  | e == eADDRINUSE = pure (Left SocketAddressInUse)+  | otherwise = die+      ("Socket.Stream.Unix.bindListen: " ++ describeErrorCode e)++describeErrorCode :: Errno -> String+describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"++-- These are the exceptions that can happen as a result+-- of calling @socket@ with the intent of using the socket+-- to listen for inbound connections.+handleSocketListenException :: Errno -> IO (Either SocketException a)+handleSocketListenException e@(Errno n)+  | e == eMFILE = pure (Left SocketFileDescriptorLimit)+  | e == eNFILE = pure (Left SocketFileDescriptorLimit)+  | otherwise = die+      ("Socket.Stream.Unix.listen: " ++ D.string e ++ " (" ++ show n ++ ")")++-- | Open a socket that is used to listen for inbound connections.+withListener ::+     UnixAddress+  -> (Listener -> IO a)+  -> IO (Either SocketException a)+withListener !endpoint f = mask $ \restore -> do+  listen endpoint >>= \case+    Left err -> pure (Left err)+    Right sck -> do+      a <- onException+        (restore (f sck))+        (unlisten_ sck)+      unlisten sck+      pure (Right a)++-- | Close a listener. This throws an unrecoverable exception if+--   the socket cannot be closed.+unlisten :: Listener -> IO ()+unlisten (Listener fd) = S.uninterruptibleClose fd >>= \case+  Left _ -> die "Socket.Stream.Unix.unlisten"+  Right _ -> pure ()++-- | Close a listener. This does not check to see whether or not+-- the operating system successfully closed the socket. It never+-- throws exceptions of any kind. This should only be preferred+-- to 'unlistener' in exception-cleanup contexts where there is+-- already an exception that will be rethrown. See the implementation+-- of 'withListener' for an example of appropriate use of both+-- 'unlistener' and 'unlistener_'.+unlisten_ :: Listener -> IO ()+unlisten_ (Listener fd) = S.uninterruptibleErrorlessClose fd++-- | Listen for an inbound connection.+accept :: Listener -> IO (Either (AcceptException 'Uninterruptible) Connection)+accept (Listener !fd) = do+  -- Although this function must be called in a context where+  -- exceptions are masked, recall that EM.wait uses an STM+  -- action that might retry, meaning that this first part is+  -- still interruptible. This is a good thing in the case of+  -- this function.+  let !mngr = EM.manager+  -- The listener should already be registered, so we can just+  -- ask for the reader directly.+  !tv <- EM.reader mngr fd+  let go !oldToken = do+        waitlessAccept fd >>= \case+          Left merr -> case merr of+            Nothing -> EM.unreadyAndWait oldToken tv >>= go+            Just err -> pure (Left err)+          Right r@(Connection conn) -> do+            EM.register mngr conn+            pure (Right r)+  go =<< STM.readTVarIO tv++-- We use the maybe to mean that the user needs to wait again.+waitlessAccept :: Fd -> IO (Either (Maybe (AcceptException i)) Connection)+waitlessAccept lstn = do+  L.uninterruptibleAccept4 lstn 200 (L.closeOnExec <> L.nonblocking) >>= \case+    Left err -> handleAcceptException err+    Right (_,_,acpt) -> pure (Right (Connection acpt))++-- These are the exceptions that can happen as a result+-- of calling @accept@.+handleAcceptException :: Errno -> IO (Either (Maybe (AcceptException i)) a)+handleAcceptException e+  | e == eAGAIN = pure (Left Nothing)+  | e == eWOULDBLOCK = pure (Left Nothing)+  | e == eCONNABORTED = pure (Left (Just AcceptConnectionAborted))+  | e == eMFILE = pure (Left (Just AcceptFileDescriptorLimit))+  | e == eNFILE = pure (Left (Just AcceptFileDescriptorLimit))+  | e == ePERM = pure (Left (Just AcceptFirewalled))+  | otherwise = die ("Socket.Stream.IPv4.accept: " ++ describeErrorCode e)++-- | 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+-- resets need to be sent.+disconnect :: Connection -> IO (Either CloseException ())+disconnect (Connection fd) = gracefulCloseA fd++gracefulCloseA :: Fd -> IO (Either CloseException ())+gracefulCloseA fd = do+  S.uninterruptibleShutdown fd S.write >>= \case+    -- On Linux (not sure about others), calling shutdown+    -- on the write channel fails with with ENOTCONN if the+    -- write channel is already closed. It is common for this to+    -- happen (e.g. if the peer calls @close@ before the local+    -- process runs gracefulClose, the local operating system+    -- will have already closed the write channel). However,+    -- it does not pose a problem. We just proceed as we would+    -- have since either way we become certain that the write channel+    -- is closed.+    Left err -> if err == eNOTCONN+      then gracefulCloseB fd+      else do+        S.uninterruptibleErrorlessClose fd+        die "Socket.Stream.Unix.gracefulCloseA"+    Right _ -> gracefulCloseB fd++gracefulCloseB :: Fd -> IO (Either CloseException ())+gracefulCloseB !fd = do+  !buf <- PM.newByteArray 1+  -- We do not actually want to remove the bytes from the+  -- receive buffer, so we use MSG_PEEK. We are certain+  -- to send a reset when a CloseException is reported.+  -- Retrospective: Why is MSG_PEEK important? Who cares if the+  -- bytes get eaten? The receive buffer is about to get axed anyway.+  S.uninterruptibleReceiveMutableByteArray fd buf 0 1 S.peek >>= \case+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN || err1 == eNOTCONN+      then do+        _ <- S.uninterruptibleClose fd+        pure (Right ())+      else do+        _ <- S.uninterruptibleClose fd+        -- We treat all @recv@ errors except for the nonblocking+        -- notices as unrecoverable.+        die "Socket.Stream.Unix.gracefulCloseB"+    Right sz -> if sz == 0+      then S.uninterruptibleClose fd >>= \case+        Left _ -> die "Socket.Stream.Unix.gracefulCloseB"+        Right _ -> pure (Right ())+      else do+        _ <- S.uninterruptibleClose fd+        pure (Left ClosePeerContinuedSending)++-- | Close a connection. This does not check to see whether or not+-- the connection was brought down gracefully. It just calls @close@+-- and is likely to cause a TCP reset to be sent. It never+-- throws exceptions of any kind (even if @close@ fails).+-- This should only be preferred+-- to 'disconnect' in exception-cleanup contexts where there is+-- already an exception that will be rethrown. See the implementation+-- of 'withConnection' for an example of appropriate use of both+-- 'disconnect' and 'disconnect_'.+disconnect_ :: Connection -> IO ()+disconnect_ (Connection fd) = S.uninterruptibleErrorlessClose fd++-- | Accept a connection on the listener and run the supplied callback+-- on it. This closes the connection when the callback finishes or if+-- an exception is thrown. Since this function blocks the thread until+-- the callback finishes, it is only suitable for stream socket clients+-- that handle one connection at a time. The variant 'forkAcceptedUnmasked'+-- is preferrable for servers that need to handle connections concurrently+-- (most use cases).+withAccepted ::+     Listener+  -> (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 (AcceptException 'Uninterruptible) b)+withAccepted !lstn consumeException cb = do+  r <- mask $ \restore -> do+    accept lstn >>= \case+      Left e -> pure (Left e)+      Right conn -> do+        a <- onException (restore (cb conn)) (disconnect_ conn)+        e <- disconnect conn+        pure (Right (e,a))+  -- Notice that consumeException gets run in an unmasked context.+  case r of+    Left e -> pure (Left e)+    Right (e,a) -> fmap Right (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+-- callback. Typically, @a@ is instantiated to @()@.+interruptibleForkAcceptedUnmasked ::+     TVar Int+     -- ^ Connection counter. Incremented when connection+     --   is accepted. Decremented after connection is closed.+  -> TVar Bool+     -- ^ Interrupted. If this becomes 'True' give up and return+     --   @'Left' 'AcceptInterrupted'@.+  -> Listener+     -- ^ Connection listener+  -> (Either CloseException () -> a -> IO ())+     -- ^ Callback to handle an ungraceful close. This must not+     --   throw an exception.+  -> (Connection -> IO a)+     -- ^ Callback to consume connection. Must not return the connection.+  -> IO (Either (AcceptException 'Interruptible) ThreadId)+interruptibleForkAcceptedUnmasked !counter !abandon !lstn consumeException cb =+  mask_ $ interruptibleAcceptCounting counter abandon lstn >>= \case+    Left e -> pure (Left e)+    Right conn -> fmap Right $ forkIOWithUnmask $ \unmask -> do+      a <- onException+        (unmask (cb conn))+        (disconnect_ conn *> atomically (modifyTVar' counter (subtract 1)))+      e <- disconnect conn+      r <- unmask (consumeException e a)+      atomically (modifyTVar' counter (subtract 1))+      pure r++-- Only used internally+interruptibleAcceptCounting :: +     TVar Int+  -> TVar Bool+  -> Listener+  -> IO (Either (AcceptException 'Interruptible) Connection)+interruptibleAcceptCounting !counter !abandon (Listener !fd) = do+  -- TODO: pull these out of the loop+  let !mngr = EM.manager+  tv <- EM.reader mngr fd+  token <- EM.interruptibleWaitCounting counter abandon tv+  if EM.isInterrupt token+    then pure (Left AcceptInterrupted)+    else waitlessAccept fd >>= \case+      Left merr -> case merr of+        Nothing -> do+          EM.unready token tv+          -- Decrement the connection counter if the notification+          -- from epoll was a false alarm.+          atomically (modifyTVar' counter (subtract 1))+          interruptibleAcceptCounting counter abandon (Listener fd)+        Just err -> pure (Left err)+      Right r@(Connection conn) -> do+        EM.register mngr conn+        pure (Right r)++-- | Retrieve a listener that systemd has passed to the process. This+-- may only be called once. There is no bracketed variant+-- of this function because the listener is expected to remain open for+-- the remainder of the application.+--+-- There are several reasons this function may return @Left@:+--+-- * @sd_listen_fds@ returned more than one file descriptor+-- * @sd_is_socket@ found that the file descriptor was not a socket or+--   that it was a socket that was not in listening mode.+systemdListener :: IO (Either SystemdException Listener)+systemdListener = coerce (systemdListenerInternal S.Unix)
+ src/Socket/Systemd.hs view
@@ -0,0 +1,54 @@+{-# language BangPatterns #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# LANGUAGE LambdaCase #-}++module Socket.Systemd+  ( SystemdException(..)+  , systemdListenerInternal+  ) where++import Control.Exception (Exception)+import Data.Bits ((.&.))+import Foreign.C.Types (CInt)+import Foreign.C.Error (Errno(..))+import System.Posix.Types (Fd(Fd))++import qualified Posix.File as F+import qualified Socket.EventManager as EM+import qualified Posix.Socket as S+import qualified Linux.Systemd as L++-- Internal function used for inheriting either an internet+-- stream socket or a unix-domain stream socket from systemd.+systemdListenerInternal :: S.Family -> IO (Either SystemdException Fd)+{-# noinline systemdListenerInternal #-}+systemdListenerInternal !fam = L.listenFds 1 >>= \case+  Left (Errno e) -> pure (Left (SystemdErrno e))+  Right n -> case n of+    1 -> L.isSocket fd0 fam S.stream 1 >>= \case+      Left (Errno e) -> pure (Left (SystemdErrno e))+      Right r -> case r of+        0 -> pure (Left SystemdDescriptorInfo)+        _ -> F.uninterruptibleGetStatusFlags fd0 >>= \case+          Left (Errno e) -> pure (Left (SystemdFnctlErrno e))+          Right status -> if F.nonblocking .&. status == mempty+            then pure (Left SystemdBlocking)+            else do+              let !mngr = EM.manager+              EM.register mngr fd0+              pure (Right fd0)+    _ -> pure (Left (SystemdDescriptorCount n))+  where+  fd0 = Fd 3++data SystemdException+  = SystemdDescriptorCount !CInt+  | SystemdDescriptorInfo+  | SystemdBlocking+    -- ^ The socket was in blocking mode. Set @NonBlocking=True@ in the systemd+    -- service to resolve this.+  | SystemdErrno !CInt+  | SystemdFnctlErrno !CInt+  deriving stock (Show,Eq)+  deriving anyclass (Exception)
test/Main.hs view
@@ -29,12 +29,16 @@ import Test.Tasty import Test.Tasty.HUnit +import qualified Data.Bytes as Bytes import qualified Data.ByteString as B import qualified Data.Primitive as PM import qualified Data.Primitive.Unlifted.Array as PM import qualified Data.Primitive.MVar as PM import qualified GHC.Exts as E import qualified Net.IPv4 as IPv4+import qualified Socket.SequencedPacket.Unix as SP+import qualified Socket.SequencedPacket.Uninterruptible.Bytes as SP+import qualified Socket.Datagram.Unix.Connected as DUC import qualified Socket.Datagram.IPv4.Unconnected as DIU import qualified Socket.Datagram.IPv4.Connected as DIC import qualified Socket.Datagram.Uninterruptible.Bytes as DUB@@ -66,7 +70,18 @@         , testCase "B" testDatagramConnectedB         ]       ]+    , testGroup "unix"+      [ testGroup "connected"+        [ testCase "C" testDatagramConnectedC+        ]+      ]     ]+  , testGroup "seqpacket"+    [ testGroup "unix"+      [ testCase "A" testSeqpacketA+      , testCase "B" testSeqpacketB+      ]+    ]   , testGroup "stream"     [ testGroup "ipv4"       [ testCase "A" testStreamA@@ -140,6 +155,52 @@       Right _ -> fail "testDatagramUndestinedB: received datagram"     ) +-- This checks that unix-domain datagram socketpair works.+testDatagramConnectedC :: Assertion+testDatagramConnectedC = do+  ((),actual) <- either throwIO pure =<< DUC.withPair+    (\src dst -> concurrently+      ( unhandled $ DUB.send src (unsliced message) )+      ( unhandled $ DUB.receive dst sz )+    )+  message @=? actual+  where+  message = E.fromList [0,1,2,3] :: ByteArray+  sz = PM.sizeofByteArray message++-- This checks that unix-domain seqpacket socketpair works.+testSeqpacketA :: Assertion+testSeqpacketA = do+  ((),actual) <- either throwIO pure =<< SP.withPair+    (\src dst -> concurrently+      ( unhandled $ SP.send src (unsliced message) )+      ( unhandled $ SP.receive dst (sz + 1) )+    )+  message @=? actual+  where+  message = E.fromList [0,1,2,3] :: ByteArray+  sz = PM.sizeofByteArray message++testSeqpacketB :: Assertion+testSeqpacketB = do+  (m :: PM.MVar RealWorld ()) <- PM.newEmptyMVar+  ((),received) <- concurrently (sender m) (receiver m)+  received @=? message+  where+  -- Socket in abstract namespace+  addr = DUC.UnixAddress (E.fromList [0,97,98,99,100,101])+  message = E.fromList (enumFromTo 0 (100 :: Word8)) :: ByteArray+  sender :: PM.MVar RealWorld () -> IO ()+  sender m = do+    PM.takeMVar m+    unhandled $ SP.withConnection addr unhandledClose $ \conn -> do+      unhandled $ SP.send conn (Bytes.fromByteArray message)+  receiver :: PM.MVar RealWorld () -> IO ByteArray+  receiver m = unhandled $ SP.withListener addr $ \listener -> do+    PM.putMVar m ()+    unhandled $ SP.withAccepted listener unhandledClose $ \conn -> do+      unhandled $ SP.receive conn 256+ testDatagramUndestinedA :: Assertion testDatagramUndestinedA = do   (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar@@ -180,8 +241,9 @@   receiver m n = unhandled $ DIU.withSocket (DIU.Peer IPv4.loopback 0) $ \sock port -> do     PM.putMVar m port     PM.takeMVar n-    slab <- DUB.newPeerlessSlab 2 (max sz1 sz2)-    msgs <- unhandled $ DUB.receiveMany sock slab+    slab <- DUB.newPeerlessSlab 2+    let sz = max sz1 sz2+    msgs <- unhandled $ DUB.receiveMany sock slab sz     let msgCount = PM.sizeofUnliftedArray msgs     if msgCount == 2       then pure (PM.indexUnliftedArray msgs 0, PM.indexUnliftedArray msgs 1)@@ -222,13 +284,14 @@   receiver m n = unhandled $ DIU.withSocket (DIU.Peer IPv4.loopback 0) $ \sock port -> do     PM.putMVar m port     PM.takeMVar n-    slab <- DUB.newIPv4Slab 3 (max sz1 (max sz2 (max sz3 sz4)))-    msgsX <- unhandled $ DUB.receiveManyFromIPv4 sock slab+    let sz = max sz1 (max sz2 (max sz3 sz4))+    slab <- DUB.newIPv4Slab 3+    msgsX <- unhandled $ DUB.receiveManyFromIPv4 sock slab sz     let msgCountX = PM.sizeofSmallArray msgsX     (msg1,msg2,msg3) <- if PM.sizeofSmallArray msgsX == 3       then pure (PM.indexSmallArray msgsX 0, PM.indexSmallArray msgsX 1, PM.indexSmallArray msgsX 2)       else fail $ "received a number of messages other than 3: " ++ show msgCountX-    msgsY <- unhandled $ DUB.receiveManyFromIPv4 sock slab+    msgsY <- unhandled $ DUB.receiveManyFromIPv4 sock slab sz     let msgCountY = PM.sizeofSmallArray msgsY     msg4 <- if msgCountY == 1       then pure (PM.indexSmallArray msgsY 0)@@ -287,12 +350,13 @@   receiver m n = unhandled $ DIU.withSocket (DIU.Peer IPv4.loopback 0) $ \sock port -> do     PM.putMVar m port     PM.takeMVar n-    slab <- DUB.newPeerlessSlab 1 (max sz1 (max sz2 sz3))-    msgsX <- unhandled $ DUB.receiveMany sock slab+    slab <- DUB.newPeerlessSlab 1+    let sz = max sz1 (max sz2 sz3)+    msgsX <- unhandled $ DUB.receiveMany sock slab sz     when (PM.sizeofUnliftedArray msgsX /= 1) $ fail "more than one message for X"-    msgsY <- unhandled $ DUB.receiveMany sock slab+    msgsY <- unhandled $ DUB.receiveMany sock slab sz     when (PM.sizeofUnliftedArray msgsX /= 1) $ fail "more than one message for Y"-    msgsZ <- unhandled $ DUB.receiveMany sock slab+    msgsZ <- unhandled $ DUB.receiveMany sock slab sz     when (PM.sizeofUnliftedArray msgsX /= 1) $ fail "more than one message for Z"     pure (PM.indexUnliftedArray msgsX 0,PM.indexUnliftedArray msgsY 0,PM.indexUnliftedArray msgsZ 0) @@ -310,8 +374,8 @@     unhandled $ DUB.sendToIPv4 sock (DIU.Peer IPv4.loopback dstPort) (unsliced message)   receiver m = unhandled $ DIU.withSocket (DIU.Peer IPv4.loopback 0) $ \sock port -> do     PM.putMVar m port-    slab <- DUB.newPeerlessSlab 1 (sz - 1)-    DUB.receiveMany sock slab+    slab <- DUB.newPeerlessSlab 1+    DUB.receiveMany sock slab (sz - 1)  testDatagramUndestinedG :: Assertion testDatagramUndestinedG = do