From 375921e57036237878e982dfbb95351850b267b9 Mon Sep 17 00:00:00 2001 From: Kazu Yamamoto Date: Mon, 21 Sep 2026 20:51:59 +0900 Subject: [PATCH] Deliver a receive through STM, and stop pretending readiness works on Windows waitReadSocketSTM is threadWaitReadSTM on the raw socket. GHC compiles the event manager branch of threadWaitRead out on Windows, so it falls back to the waitRead# primop, which the threaded RTS does not implement for sockets. The STM action therefore never fires -- under the old I/O manager as much as under WinIO, so this is not a WinIO regression. It blocks forever and says nothing. Readiness cannot be recovered on Windows in general: completion ports report that an operation finished, not that one could be started. So offer the completion shape instead. recvBufSTM and recvBufFromSTM, and recvSTM and recvFromSTM for ByteStrings, start a receive and hand its result over through STM. They work on every platform, and on Windows they are what callers who used waitReadSocketSTM want. The four readiness functions are now POSIX only and throw on Windows, naming what to do instead: use one of the receives above, or, for the write side, just issue the send -- completion ports give no way to ask whether one would block and none is needed. Cancelling a receive can lose data that the kernel has already dequeued, and needs the receive to be interruptible, which the old Windows I/O manager does not provide. Both are documented, and the cancellation test is POSIX only for the second reason. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 4 + Network/Socket/ByteString.hs | 2 + Network/Socket/ByteString/IO.hsc | 21 +++++ Network/Socket/STM.hs | 121 ++++++++++++++++++++++++- network.cabal | 1 + tests/Network/Socket/ByteStringSpec.hs | 37 ++++++++ 6 files changed, 183 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f41b8c7a..d9f4e58f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,10 @@ [#626](https://github.com/haskell/network/pull/626) * WINIO: fixing a spurious EOF when a recv completes synchronously. [#626](https://github.com/haskell/network/pull/626) +* Adding recvSTM, recvFromSTM, recvBufSTM and recvBufFromSTM, which + deliver a receive through STM and work on Windows too. The readiness + based waitReadSocketSTM and waitWriteSocketSTM cannot work there and + now say so instead of blocking forever. ## Version 3.2.9.0 diff --git a/Network/Socket/ByteString.hs b/Network/Socket/ByteString.hs index 42424e94..a9b943c2 100644 --- a/Network/Socket/ByteString.hs +++ b/Network/Socket/ByteString.hs @@ -32,6 +32,8 @@ module Network.Socket.ByteString ( -- * Receive data from a socket recv, recvFrom, + recvSTM, + recvFromSTM, -- * Advanced send and recv sendMsg, diff --git a/Network/Socket/ByteString/IO.hsc b/Network/Socket/ByteString/IO.hsc index a600b26c..4aed83ea 100644 --- a/Network/Socket/ByteString/IO.hsc +++ b/Network/Socket/ByteString/IO.hsc @@ -29,6 +29,8 @@ module Network.Socket.ByteString.IO -- * Receive data from a socket , recv , recvFrom + , recvSTM + , recvFromSTM , waitWhen0 -- * Advanced send and recv @@ -58,6 +60,8 @@ import System.Posix.Types (Fd(..)) import Network.Socket.Flag import Network.Socket.SockAddr (annotateWithSocket) +import Network.Socket.STM (viaSTM) +import Control.Concurrent.STM (STM) #if defined(mingw32_HOST_OS) import Network.Socket.Win32.Load #endif @@ -356,3 +360,20 @@ recvMsg s siz clen flags = recvMsg' `annotateWithSocket` (s, Nothing) let bs' | len < siz = PS fptr 0 len | otherwise = bs return (addr, bs', cmsgs, flags') + +-- | Start receiving and return an 'STM' action delivering the data, +-- together with an action cancelling the receive. Unlike +-- 'Network.Socket.waitReadSocketSTM' this works on Windows too, +-- because it waits for the receive to complete rather than for the +-- socket to become readable. +-- +-- Cancelling can lose data: the receive may already have taken a +-- datagram out of the kernel queue. Only abandon the 'STM' action +-- where losing it is acceptable, such as on shutdown. +recvSTM :: Socket -> Int -> IO (STM ByteString, IO ()) +recvSTM s nbytes = viaSTM $ recv s nbytes + +-- | 'recvSTM' for unconnected sockets, also returning the peer address. +-- The same caveat about cancelling applies. +recvFromSTM :: Socket -> Int -> IO (STM (ByteString, SockAddr), IO ()) +recvFromSTM s nbytes = viaSTM $ recvFrom s nbytes diff --git a/Network/Socket/STM.hs b/Network/Socket/STM.hs index 6f2419b4..656326ae 100644 --- a/Network/Socket/STM.hs +++ b/Network/Socket/STM.hs @@ -1,24 +1,139 @@ -module Network.Socket.STM where +{-# LANGUAGE CPP #-} +{-# LANGUAGE LambdaCase #-} -import Control.Concurrent +-- | STM interfaces for sockets. +-- +-- Two styles are offered here: +-- +-- * Readiness: 'waitReadSocketSTM' and friends return an 'STM' action +-- that becomes available once the socket is ready. This mirrors +-- @select@\/@epoll@ and is __POSIX only__: Windows completion ports +-- report that an operation has finished, not that one could be +-- started, so there is no readiness to wait for. All four throw on +-- Windows. +-- +-- * Completion: 'recvBufSTM' and 'recvBufFromSTM' start a receive and +-- return an 'STM' action that delivers its result. These work on +-- every platform and are what you want if the code has to run on +-- Windows. "Network.Socket.ByteString" has 'ByteString' versions. +module Network.Socket.STM ( + -- * Waiting for readiness (POSIX only) + waitReadSocketSTM, + waitAndCancelReadSocketSTM, + waitWriteSocketSTM, + waitAndCancelWriteSocketSTM, + + -- * Receiving through STM (all platforms) + recvBufSTM, + recvBufFromSTM, + + -- * Building block + viaSTM, +) where + +import Control.Concurrent (forkIO, killThread) import Control.Concurrent.STM +import qualified Control.Exception as E +import Network.Socket.Buffer +import Network.Socket.Imports import Network.Socket.Types -import System.Posix.Types +#if !defined(mingw32_HOST_OS) +import Control.Concurrent (threadWaitReadSTM, threadWaitWriteSTM) +import System.Posix.Types (Fd (..)) +#endif -- | STM action to wait until the socket is ready for reading. +-- +-- __POSIX only.__ On Windows this throws: completion ports have no +-- notion of readiness. Use 'recvBufFromSTM', or +-- @Network.Socket.ByteString.recvFromSTM@, instead. waitReadSocketSTM :: Socket -> IO (STM ()) waitReadSocketSTM s = fst <$> waitAndCancelReadSocketSTM s -- | STM action to wait until the socket is ready for reading and STM -- action to cancel the waiting. +-- +-- __POSIX only.__ See 'waitReadSocketSTM'. waitAndCancelReadSocketSTM :: Socket -> IO (STM (), IO ()) +#if defined(mingw32_HOST_OS) +waitAndCancelReadSocketSTM _ = + ioError $ + userError $ + "waitAndCancelReadSocketSTM: Windows completion ports do not " + ++ "provide readiness notification; use recvBufFromSTM or " + ++ "Network.Socket.ByteString.recvFromSTM instead" +#else waitAndCancelReadSocketSTM s = withFdSocket s $ threadWaitReadSTM . Fd . fromIntegral +#endif -- | STM action to wait until the socket is ready for writing. +-- +-- __POSIX only.__ On Windows this throws. Completion ports give no +-- way to ask whether a send would block, and none is needed: issue +-- the send and let it complete asynchronously. waitWriteSocketSTM :: Socket -> IO (STM ()) waitWriteSocketSTM s = fst <$> waitAndCancelWriteSocketSTM s -- | STM action to wait until the socket is ready for writing and STM -- action to cancel the waiting. +-- +-- __POSIX only.__ See 'waitWriteSocketSTM'. waitAndCancelWriteSocketSTM :: Socket -> IO (STM (), IO ()) +#if defined(mingw32_HOST_OS) +waitAndCancelWriteSocketSTM _ = + ioError $ + userError $ + "waitAndCancelWriteSocketSTM: Windows completion ports do not " + ++ "provide readiness notification, and none is needed for " + ++ "sending: issue the send instead" +#else waitAndCancelWriteSocketSTM s = withFdSocket s $ threadWaitWriteSTM . Fd . fromIntegral +#endif + +-- | Start receiving into the given buffer and return an 'STM' action +-- delivering the number of bytes received, together with an action +-- cancelling the receive. If the receive fails, the 'STM' action +-- rethrows the exception. +-- +-- Unlike 'waitReadSocketSTM' this works on Windows, because it waits +-- for a completion rather than for readiness. +-- +-- Two consequences follow from that, and both matter when composing +-- with 'orElse': +-- +-- * Cancelling can lose data. The receive may already have taken a +-- datagram out of the kernel queue, and that datagram is then gone. +-- Only abandon the 'STM' action on paths where losing it is +-- acceptable, such as shutdown. +-- +-- * Cancelling needs the receive to be interruptible. It is on POSIX +-- and under WinIO, but not under the old Windows I/O manager, where +-- the receive blocks in a foreign call and the cancel action waits +-- for it. +-- +-- * The buffer must stay alive until the 'STM' action completes or +-- the cancel action returns. +recvBufSTM :: Socket -> Ptr Word8 -> Int -> IO (STM Int, IO ()) +recvBufSTM s ptr nbytes = viaSTM $ recvBuf s ptr nbytes + +-- | 'recvBufSTM' for unconnected sockets, also returning the peer +-- address. The same caveats apply. +recvBufFromSTM + :: SocketAddress sa => Socket -> Ptr Word8 -> Int -> IO (STM (Int, sa), IO ()) +recvBufFromSTM s ptr nbytes = viaSTM $ recvBufFrom s ptr nbytes + +-- | Run a blocking socket operation in a separate thread and hand its +-- result over through STM, together with an action cancelling it. +-- This is what 'recvBufSTM' is built from; the same caveats about +-- cancelling and about buffer lifetime apply to anything built with +-- it. +viaSTM :: IO a -> IO (STM a, IO ()) +viaSTM act = do + var <- newTVarIO Nothing + tid <- forkIO $ E.try act >>= atomically . writeTVar var . Just + let wait = + readTVar var >>= \case + Nothing -> retry + Just (Left e) -> throwSTM (e :: E.SomeException) + Just (Right x) -> return x + return (wait, killThread tid) diff --git a/network.cabal b/network.cabal index 675775c5..02d6825e 100644 --- a/network.cabal +++ b/network.cabal @@ -204,6 +204,7 @@ test-suite spec directory, HUnit, network, + stm, temporary, hspec >=2.6, QuickCheck diff --git a/tests/Network/Socket/ByteStringSpec.hs b/tests/Network/Socket/ByteStringSpec.hs index ac5801df..1e962731 100644 --- a/tests/Network/Socket/ByteStringSpec.hs +++ b/tests/Network/Socket/ByteStringSpec.hs @@ -1,8 +1,11 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} module Network.Socket.ByteStringSpec (main, spec) where +import Control.Concurrent (threadDelay) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Concurrent.STM (atomically) import Control.Monad import Data.Bits import qualified Data.ByteString as S @@ -21,6 +24,40 @@ main = hspec spec spec :: Spec spec = do + describe "recvSTM" $ do + it "delivers the data through STM" $ do + let server sock = do + (wait, _cancel) <- recvSTM sock 1024 + atomically wait `shouldReturn` testMsg + client sock = send sock testMsg + tcpTest client server + +#if !defined(mingw32_HOST_OS) + -- Cancelling needs the underlying receive to be interruptible. + -- It is on POSIX and under WinIO, but not under the old Windows + -- I/O manager, where the receive blocks in a foreign call that + -- killThread cannot reach. + it "can be cancelled" $ do + let server sock = do + (_wait, cancel) <- recvSTM sock 1024 + cancel + -- the socket is still usable afterwards + recv sock 1024 `shouldReturn` testMsg + client sock = do + threadDelay 100000 + void $ send sock testMsg + tcpTest client server +#endif + + describe "recvFromSTM" $ do + it "delivers the datagram and the peer address through STM" $ do + let server sock = do + (wait, _cancel) <- recvFromSTM sock 1024 + (bs, _) <- atomically wait + bs `shouldBe` testMsg + client sock serverAddr' = void $ sendTo sock testMsg serverAddr' + udpTest client server + describe "send" $ do it "works well" $ do let server sock = recv sock 1024 `shouldReturn` testMsg