sockets 0.6.1.1 → 0.7.0.0
raw patch · 13 files changed
+56/−180 lines, 13 filesdep +systemd-apidep ~posix-apidep ~primitive-unlifted
Dependencies added: systemd-api
Dependency ranges changed: posix-api, primitive-unlifted
Files
- CHANGELOG.md +6/−0
- sockets.cabal +14/−13
- src-internal/Socket/Bytes.hs +1/−1
- src-internal/Socket/Connected/Bytes.hs +1/−1
- src-internal/Socket/Connected/MutableBytes.hs +1/−1
- src-internal/Socket/Destined/IPv4/Bytes.hs +1/−1
- src-internal/Socket/Destined/IPv4/MutableBytes.hs +1/−1
- src-internal/Socket/EventManager.hs +28/−18
- src-internal/Socket/MutableBytes.hs +2/−2
- src-internal/Socket/MutableBytes/Peerless.hs +1/−1
- src/Socket/Stream/Uninterruptible/Bytes.hs +0/−59
- src/Socket/Stream/Uninterruptible/MutableBytes.hs +0/−18
- test/Main.hs +0/−64
CHANGELOG.md view
@@ -1,5 +1,11 @@ # Revision history for stream-sockets +## 0.7.0.0++* Remove `sendMany`.+* Make it possible to build this project with GHC 9.2 series. The previous+ release had dropped support for everything below GHC 9.4.+ ## 0.6.1.1 * Work with primitive-unlifted-2.1.0.0.
sockets.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: sockets-version: 0.6.1.1+version: 0.7.0.0 synopsis: High-level network sockets description: This library provides a high-level abstraction for network sockets. It uses@@ -83,9 +83,9 @@ , base >= 4.11.1.0 && <5 , bytestring >=0.10 && <0.12 , ip >= 1.4.1- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , primitive-offset >= 0.2 && <0.3- , primitive-unlifted >= 2.1+ , primitive-unlifted ==0.1.3.1 || >= 2.1 , primitive-addr >= 0.1 && <0.2 , stm >= 2.4 , text >= 1.2@@ -156,7 +156,7 @@ build-depends: , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , stm >= 2.4 , sockets-internal , sockets-buffer@@ -174,7 +174,7 @@ , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2 , stm >= 2.4- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , primitive-offset >= 0.2 && <0.3 , sockets-internal , sockets-buffer@@ -195,8 +195,8 @@ , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2 , stm >= 2.4- , posix-api >=0.4.0.1 && <0.5- , primitive-unlifted >= 2.1+ , posix-api >=0.7 && <0.8+ , primitive-unlifted ==0.1.3.1 || >= 2.1 , byteslice >= 0.1.1 , sockets-datagram-receive , sockets-internal@@ -219,7 +219,7 @@ build-depends: , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , stm >= 2.4 , sockets-internal , sockets-buffer@@ -237,7 +237,7 @@ build-depends: , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , stm >= 2.4 , sockets-internal , sockets-buffer@@ -255,7 +255,7 @@ build-depends: , base >= 4.11.1.0 && <5 , error-codes >= 0.1.0.1 && < 0.2- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8 , stm >= 2.4 , sockets-internal , sockets-interrupt@@ -532,10 +532,11 @@ , bytestring >=0.10 && <0.12 , error-codes >=0.1.0.1 && <0.2 , ip >=1.4.1- , posix-api >=0.4.0.1 && <0.5+ , posix-api >=0.7 && <0.8+ , systemd-api >=0.1 , primitive-addr >= 0.1 && <0.2 , primitive-offset >= 0.2 && <0.3- , primitive-unlifted >= 2.1+ , primitive-unlifted ==0.1.3.1 || >= 2.1 , sockets-datagram-receive , sockets-datagram-receive-many , sockets-datagram-send@@ -570,7 +571,7 @@ , async , bytestring , byteslice >= 0.1.1- , primitive-unlifted >= 2.1+ , primitive-unlifted ==0.1.3.1 || >= 2.1 , primitive-addr , stm if flag(checked)
src-internal/Socket/Bytes.hs view
@@ -37,5 +37,5 @@ sendOnce :: Fd -> Bytes -> IO (Either Errno CSize) {-# inline sendOnce #-} sendOnce fd (Bytes arr off len) =- uninterruptibleSendByteArray fd arr (intToCInt off) (intToCSize len) noSignal+ uninterruptibleSendByteArray fd arr off (intToCSize len) noSignal
src-internal/Socket/Connected/Bytes.hs view
@@ -26,7 +26,7 @@ -- No need for MSG_NOSIGNAL since this is a datagram -- socket, not a stream socket. S.uninterruptibleSendByteArray sock arr- (intToCInt off)+ off (intToCSize len) mempty
src-internal/Socket/Connected/MutableBytes.hs view
@@ -26,7 +26,7 @@ -- No need for MSG_NOSIGNAL since this is a datagram -- socket, not a stream socket. S.uninterruptibleSendMutableByteArray sock arr- (intToCInt off)+ off (intToCSize len) mempty
src-internal/Socket/Destined/IPv4/Bytes.hs view
@@ -26,7 +26,7 @@ send :: Peer -> Fd -> Buffer -> IO (Either Errno CSize) send !dst !sock (Bytes arr off len) = S.uninterruptibleSendToInternetByteArray sock arr- (intToCInt off)+ off (intToCSize len) mempty (endpointToSocketAddressInternet dst)
src-internal/Socket/Destined/IPv4/MutableBytes.hs view
@@ -27,7 +27,7 @@ send :: Peer -> Fd -> Buffer -> IO (Either Errno CSize) send !dst !sock (MutableBytes arr off len) = S.uninterruptibleSendToInternetMutableByteArray sock arr- (intToCInt off)+ off (intToCSize len) mempty (endpointToSocketAddressInternet dst)
src-internal/Socket/EventManager.hs view
@@ -37,7 +37,7 @@ import Data.Bits (countLeadingZeros,finiteBitSize,unsafeShiftL,(.|.),(.&.)) import Data.Bits (unsafeShiftR) import Data.Primitive.Unlifted.Array (MutableUnliftedArray(..))-import Data.Primitive (MutableByteArray(..),MutablePrimArray(..))+import Data.Primitive (MutableByteArray(..),MutablePrimArray(..),SmallMutableArray(..)) import Data.Primitive (Prim) import Data.Word (Word64,Word32) import Foreign.C.Error (Errno(..),eINTR)@@ -154,7 +154,7 @@ -- Precondition: A previous call to register has been made. -- unregister -type MUArray a = MutableUnliftedArray RealWorld a+type MUArray a = SmallMutableArray RealWorld a data Manager = Manager { variables :: !(MUArray (MUArray (TVar Token)))@@ -171,9 +171,9 @@ manager = unsafePerformIO $ do when (not rtsSupportsBoundThreads) $ do fail $ "Socket.Event.manager: threaded runtime required"- !novars <- PM.unsafeNewUnliftedArray 1+ !novars <- unsafeNewMUArray 1 writeTVarArray novars 0 =<< STM.newTVarIO (Token 0)- !variables <- PM.unsafeNewUnliftedArray 32+ !variables <- unsafeNewMUArray 32 let goX !ix = if ix >= 0 then do writeMutableUnliftedArrayArray variables ix novars@@ -231,7 +231,7 @@ lookupGeneric :: Int -- Read: 0, Write: 1 -> Int -- File descriptor- -> MutableUnliftedArray RealWorld (MutableUnliftedArray RealWorld (TVar Token))+ -> MUArray (MUArray (TVar Token)) -> IO (TVar Token) lookupGeneric !rw !fd !arr = do let (ixTier1,ixTier2) = decompose fd@@ -245,12 +245,12 @@ constructivelyLookupTier1 !fd Manager{variables,novars} = do let (ixTier1,ixTier2) = decompose fd varsTier2 <- readMutableUnliftedArrayArray variables ixTier1- if PM.sameMutableUnliftedArray varsTier2 novars+ if sameMUArray varsTier2 novars then do -- We want 2 * 2^N tvars because there is a separate read and -- write tvar for every file descriptor. let !len = exp2succ ixTier1- varsAttempt <- PM.unsafeNewUnliftedArray len+ varsAttempt <- unsafeNewMUArray len let goVars !ix = if ix > (-1) then do writeTVarArray varsAttempt ix@@ -270,7 +270,7 @@ 0 -> STM.writeTVar syncVar (Token 1) _ -> STM.retry varsTier2' <- readMutableUnliftedArrayArray variables ixTier1- when (PM.sameMutableUnliftedArray varsTier2' novars) $ do+ when (sameMUArray varsTier2' novars) $ do writeMutableUnliftedArrayArray variables ixTier1 varsAttempt STM.atomically (STM.writeTVar syncVar (Token 0)) -- We ignore the success of casUnliftedArray. It does not actually@@ -674,32 +674,42 @@ = PM.primitive (\s# -> case Exts.newPinnedByteArray# (n# *# PM.sizeOf# (undefined :: a)) s# of (# s'#, arr# #) -> (# s'#, MutablePrimArray arr# #)) +sameMUArray :: MUArray a -> MUArray a -> Bool+sameMUArray (SmallMutableArray x) (SmallMutableArray y) = isTrue# (Exts.sameSmallMutableArray# x y)++unsafeNewMUArray :: Int -> IO (MUArray a)+unsafeNewMUArray !n = PM.newSmallArray n implementationMistake++implementationMistake :: a+{-# noinline implementationMistake #-}+implementationMistake = errorWithoutStackTrace "Socket.EventManager: implementation mistake"+ readTVarArray :: forall a.- MutableUnliftedArray RealWorld (TVar a) -- ^ source+ MUArray (TVar a) -- ^ source -> Int -- ^ index -> IO (TVar a)-readTVarArray = PM.readUnliftedArray+readTVarArray = PM.readSmallArray readMutableUnliftedArrayArray - :: MutableUnliftedArray RealWorld (MutableUnliftedArray RealWorld a) -- ^ source+ :: MUArray (MUArray a) -- ^ source -> Int -- ^ index- -> IO (MutableUnliftedArray RealWorld a)-readMutableUnliftedArrayArray = PM.readUnliftedArray+ -> IO (MUArray a)+readMutableUnliftedArrayArray = PM.readSmallArray -- See readTVarArray writeTVarArray :: forall a.- MutableUnliftedArray RealWorld (TVar a) -- ^ destination+ MUArray (TVar a) -- ^ destination -> Int -- ^ index -> TVar a -- ^ value -> IO ()-writeTVarArray = PM.writeUnliftedArray+writeTVarArray !a !b !c = PM.writeSmallArray a b c writeMutableUnliftedArrayArray :: forall a.- MutableUnliftedArray RealWorld (MutableUnliftedArray RealWorld a) -- ^ source+ MUArray (MUArray a) -- ^ source -> Int -- ^ index- -> MutableUnliftedArray RealWorld a -- ^ value+ -> MUArray a -- ^ value -> IO ()-writeMutableUnliftedArrayArray = PM.writeUnliftedArray+writeMutableUnliftedArrayArray !a !b !c = PM.writeSmallArray a b c -- [Notes on registration] -- This interface requires every call to @register@ to be paired with
src-internal/Socket/MutableBytes.hs view
@@ -33,12 +33,12 @@ sendOnce :: Fd -> MutableBytes RealWorld -> IO (Either Errno CSize) {-# inline sendOnce #-} sendOnce fd (MutableBytes arr off len) =- uninterruptibleSendMutableByteArray fd arr (intToCInt off) (intToCSize len) noSignal+ uninterruptibleSendMutableByteArray fd arr off (intToCSize len) noSignal receiveOnce :: Fd -> MutableBytes RealWorld -> IO (Either Errno CSize) {-# inline receiveOnce #-} receiveOnce fd (MutableBytes arr off len) =- uninterruptibleReceiveMutableByteArray fd arr (intToCInt off) (intToCSize len) mempty+ uninterruptibleReceiveMutableByteArray fd arr off (intToCSize len) mempty intToCInt :: Int -> CInt {-# inline intToCInt #-}
src-internal/Socket/MutableBytes/Peerless.hs view
@@ -38,7 +38,7 @@ {-# inline receiveFromOnce #-} receiveFromOnce fd (MutableBytes arr off len) flags !_ = S.uninterruptibleReceiveFromMutableByteArray_- fd arr (intToCInt off) (intToCSize len) flags+ fd arr off (intToCSize len) flags intToCInt :: Int -> CInt {-# inline intToCInt #-}
src/Socket/Stream/Uninterruptible/Bytes.hs view
@@ -14,7 +14,6 @@ module Socket.Stream.Uninterruptible.Bytes ( -- * Send send- , sendMany -- * Receive , receiveExactly , receiveOnce@@ -106,64 +105,6 @@ marr1 <- PM.resizeMutableByteArray marr0 sz !arr <- PM.unsafeFreezeByteArray marr1 pure $! Right $! arr---- | Send many buffers with vectored I/O. This uses @sendmsg@ to send--- multiple chunks at a time. It may call @sendmsg@ more than once--- if there is not enough space in the operating system send buffer--- to house all the chunks at the same time.-sendMany ::- Connection -- ^ Connection- -> UnliftedArray ByteArray -- ^ Byte arrays- -> IO (Either (SendException 'Uninterruptible) ())-{-# inline sendMany #-}-sendMany (Connection conn) bufs = do- -- TODO: Generalize sendMany so that an interruptible variant- -- can be provided.- let !mngr = EM.manager- tv <- EM.writer mngr conn- -- Do not forget to use the correct wait function when this- -- is generalized.- token0 <- EM.wait tv- sendManyLoop conn tv token0 bufs 0 (PM.sizeofUnliftedArray bufs) 0--sendManyLoop ::- Fd -> TVar Token -> Token- -> UnliftedArray ByteArray- -> Int -> Int -> Int- -> IO (Either (SendException 'Uninterruptible) ())-sendManyLoop !conn !tv !old !bufs !chunkIx !chunkCount !chunkOff = if chunkCount > 0- then S.uninterruptibleSendByteArrays conn bufs chunkIx chunkCount chunkOff S.noSignal >>= \case- Left e ->- if | e == eAGAIN || e == eWOULDBLOCK -> do- EM.unready old tv- -- Do not forget to use the correct wait function when this- -- is generalized.- new <- EM.wait tv- sendManyLoop conn tv new bufs chunkIx chunkCount chunkOff- | e == ePIPE -> pure (Left SendShutdown)- | e == eCONNRESET -> pure (Left SendReset)- | otherwise -> die ("Socket.Stream.sendMany: " ++ describeErrorCode e)- Right sz' -> do- let sz0 = fromIntegral @CSize @Int sz'- -- We enumerate the sent chunks, one by one,- -- to compute the new chunk index.- -- ix: chunk index- -- m: offset into the current chunk- -- n: remaining bytes to account for- let go !ix !m !n = if n > 0- then- let !fullSz = PM.sizeofByteArray (PM.indexUnliftedArray bufs ix)- !sz = fullSz - m- in if sz <= n- then go (ix + 1) 0 (n - sz)- else (ix, m + n)- else (ix,0)- (nextChunkIx,nextChunkOff) = go chunkIx chunkOff sz0- sendManyLoop conn tv old bufs nextChunkIx (chunkCount - (nextChunkIx - chunkIx)) nextChunkOff- -- chunkCount must be 0 in this case since it is never negative- else do- whenDebugging $ when (chunkCount < 0) $ die "sendManyLoop: negative chunks"- pure (Right ()) describeErrorCode :: Errno -> String describeErrorCode err@(Errno e) = "error code " ++ D.string err ++ " (" ++ show e ++ ")"
src/Socket/Stream/Uninterruptible/MutableBytes.hs view
@@ -5,7 +5,6 @@ module Socket.Stream.Uninterruptible.MutableBytes ( -- * Send send- , sendMany -- * Receive , receiveExactly , receiveOnce@@ -14,7 +13,6 @@ import Data.Bytes.Types (MutableBytes) import Data.Primitive (MutableByteArray)-import Data.Primitive.Unlifted.Array (UnliftedArray,UnliftedArray_(UnliftedArray)) import GHC.Exts (RealWorld,proxy#) import Socket.Stream (Connection,ReceiveException,SendException) import Socket (Interruptibility(Uninterruptible))@@ -32,22 +30,6 @@ -> 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 (unsafeCoerce# bufs)) -- | Receive a number of bytes exactly equal to the length of the -- buffer slice. If needed, this may call @recv@ repeatedly until
test/Main.hs view
@@ -99,8 +99,6 @@ , testCase "12MB" (testStreamF (4 * 1024)) , testCase "48MB" (testStreamF (16 * 1024)) ]- , testCase "G" testStreamG- , testCase "H" testStreamH ] ] ]@@ -604,68 +602,6 @@ y <- unhandled $ UMB.receiveBetween conn (MutableBytes marr x 150) 100 unhandled $ UMB.receiveExactly conn (MutableBytes marr (x + y) (256 - (x + y))) PM.unsafeFreezeByteArray marr--testStreamG :: Assertion-testStreamG = do- (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar- ((),received) <- concurrently (sender m) (receiver m)- received @=? (mconcat [messageA, messageB, messageC, messageD])- where- messageA = E.fromList [0..17] :: ByteArray- messageB = E.fromList [18..92] :: ByteArray- messageC = E.fromList [93..182] :: ByteArray- messageD = E.fromList [183..255] :: ByteArray- sender :: PM.MVar RealWorld Word16 -> IO ()- sender m = do- dstPort <- PM.takeMVar m- unhandled $ SI.withConnection (DIU.Peer IPv4.loopback dstPort) unhandledClose $ \conn -> do- let msgs = E.fromList [messageA,messageB,messageC,messageD]- unhandled $ UB.sendMany conn msgs- receiver :: PM.MVar RealWorld Word16 -> IO ByteArray- receiver m = unhandled $ SI.withListener (SI.Peer IPv4.loopback 0) $ \listener port -> do- PM.putMVar m port- unhandled $ SI.withAccepted listener unhandledClose $ \conn _ -> do- marr <- PM.newByteArray 256- x <- unhandled $ UMB.receiveBetween conn (MutableBytes marr 0 60) 20- y <- unhandled $ UMB.receiveBetween conn (MutableBytes marr x 150) 100- unhandled $ UMB.receiveExactly conn (MutableBytes marr (x + y) (256 - (x + y)))- PM.unsafeFreezeByteArray marr---- The sender sends a large amount of traffic that may exceed--- the size of the operating system's TCP send buffer, currently--- 32MB. This uses sendmsg.-testStreamH :: Assertion-testStreamH = do- (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar- ((),()) <- concurrently (sender m) (receiver m)- pure ()- where- recvChunkSize = 32 * 1024- message = E.fromList (replicate (1024 * 256) magicByte) :: ByteArray- messages = E.fromList (replicate 128 message) :: UnliftedArray ByteArray- sender :: PM.MVar RealWorld Word16 -> IO ()- sender m = do- dstPort <- PM.takeMVar m- unhandled $ SI.withConnection (DIU.Peer IPv4.loopback dstPort) unhandledClose $ \conn -> do- unhandled $ UB.sendMany conn messages- receiver :: PM.MVar RealWorld Word16 -> IO ()- receiver m = unhandled $ SI.withListener (SI.Peer IPv4.loopback 0) $ \listener port -> do- PM.putMVar m port- unhandled $ SI.withAccepted listener unhandledClose $ \conn _ -> do- buffer <- PM.newByteArray recvChunkSize- let receiveLoop !remaining- | remaining > 0 = do- let recvSize = min remaining recvChunkSize- PM.setByteArray buffer 0 recvChunkSize (0 :: Word8)- bytesReceived <- unhandled- (UMB.receiveOnce conn (MutableBytes buffer 0 recvSize))- verifyClientSendBytes buffer bytesReceived >>= \case- True -> receiveLoop (remaining - bytesReceived)- False -> throwIO MagicByteMismatch- | remaining == 0 = pure ()- | otherwise = throwIO NegativeByteCount- receiveLoop (32 * 1024 * 1024)- pure () touchMutableByteArray :: MutableByteArray RealWorld -> IO () touchMutableByteArray (MutableByteArray x) = touchMutableByteArray# x