diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/sockets.cabal b/sockets.cabal --- a/sockets.cabal +++ b/sockets.cabal @@ -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) diff --git a/src-internal/Socket/Bytes.hs b/src-internal/Socket/Bytes.hs --- a/src-internal/Socket/Bytes.hs +++ b/src-internal/Socket/Bytes.hs @@ -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 diff --git a/src-internal/Socket/Connected/Bytes.hs b/src-internal/Socket/Connected/Bytes.hs --- a/src-internal/Socket/Connected/Bytes.hs +++ b/src-internal/Socket/Connected/Bytes.hs @@ -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 diff --git a/src-internal/Socket/Connected/MutableBytes.hs b/src-internal/Socket/Connected/MutableBytes.hs --- a/src-internal/Socket/Connected/MutableBytes.hs +++ b/src-internal/Socket/Connected/MutableBytes.hs @@ -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 diff --git a/src-internal/Socket/Destined/IPv4/Bytes.hs b/src-internal/Socket/Destined/IPv4/Bytes.hs --- a/src-internal/Socket/Destined/IPv4/Bytes.hs +++ b/src-internal/Socket/Destined/IPv4/Bytes.hs @@ -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) diff --git a/src-internal/Socket/Destined/IPv4/MutableBytes.hs b/src-internal/Socket/Destined/IPv4/MutableBytes.hs --- a/src-internal/Socket/Destined/IPv4/MutableBytes.hs +++ b/src-internal/Socket/Destined/IPv4/MutableBytes.hs @@ -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) diff --git a/src-internal/Socket/EventManager.hs b/src-internal/Socket/EventManager.hs --- a/src-internal/Socket/EventManager.hs +++ b/src-internal/Socket/EventManager.hs @@ -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 diff --git a/src-internal/Socket/MutableBytes.hs b/src-internal/Socket/MutableBytes.hs --- a/src-internal/Socket/MutableBytes.hs +++ b/src-internal/Socket/MutableBytes.hs @@ -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 #-} diff --git a/src-internal/Socket/MutableBytes/Peerless.hs b/src-internal/Socket/MutableBytes/Peerless.hs --- a/src-internal/Socket/MutableBytes/Peerless.hs +++ b/src-internal/Socket/MutableBytes/Peerless.hs @@ -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 #-} diff --git a/src/Socket/Stream/Uninterruptible/Bytes.hs b/src/Socket/Stream/Uninterruptible/Bytes.hs --- a/src/Socket/Stream/Uninterruptible/Bytes.hs +++ b/src/Socket/Stream/Uninterruptible/Bytes.hs @@ -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 ++ ")" diff --git a/src/Socket/Stream/Uninterruptible/MutableBytes.hs b/src/Socket/Stream/Uninterruptible/MutableBytes.hs --- a/src/Socket/Stream/Uninterruptible/MutableBytes.hs +++ b/src/Socket/Stream/Uninterruptible/MutableBytes.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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