diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Revision history for network-run +## 0.7.0 + +* The accept loop is stopped by being told, through the new + `settingsInstallShutdownHandler`, rather than by having the listening + socket closed under it. It accepts with `acceptWith`, which gives up + on the accept itself, so nothing has to be closed or interrupted. A + closed socket still stops the loop, now by returning rather than by + throwing. +* The UDP server's receive loop is stopped the same way, through + `recvFromWith`, rather than by an asynchronous exception. +* Breaking change: `ServerSettings` gains `settingsInstallShutdownHandler`. +* Breaking change: the TCP servers return `IO ()` rather than `IO a`, + since they can now return. +* `Listener` and `makeListener` are exported for servers with an accept + loop of their own. +* Which errno `accept` may be retried on comes from network's + `isTransientAcceptError` rather than from a list here, and this one + was the shorter of the two: `ECONNABORTED` and `EINTR` only, where + accept(2) names six more. +* Requires network 3.2.10 for `acceptWith`. + ## 0.6.0 * Fixing a bug that the `[(SocketOption, Int)]` variants diff --git a/Network/Run/Core.hs b/Network/Run/Core.hs --- a/Network/Run/Core.hs +++ b/Network/Run/Core.hs @@ -16,6 +16,8 @@ labelMe, safeAccept, safeAcceptWith, + Listener (..), + makeListener, ServerSettings (..), defaultServerSettings, gcloseWith, @@ -27,13 +29,12 @@ import Control.Arrow hiding (loop) import Control.Concurrent +import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad (void, when) import Data.List.NonEmpty (NonEmpty) -import Foreign.C.Error (Errno (..), eCONNABORTED) import Foreign.C.Types (CInt) import GHC.Conc.Sync -import GHC.IO.Exception (IOErrorType (Interrupted), ioe_errno) import Network.Socket import System.IO.Error (ioeGetErrorType, isFullError) @@ -208,6 +209,23 @@ , settingsAcceptRetryDelay :: Int -- ^ Microseconds to wait before retrying 'accept' after running -- out of file descriptors. The default is 100000. + , settingsInstallShutdownHandler :: IO () -> IO () + -- ^ An action to install a handler (a Unix signal handler, say) which + -- stops the server. The first argument is the action which does that: + -- it stops the loop accepting, and the listening socket is closed + -- afterwards by whoever opened it. It is not an action which closes + -- that socket, and should not be used as one. The default does + -- nothing, and the server is then stopped the other way, by an + -- asynchronous exception to the thread running it. + -- + -- __The server returns without waiting for the connections it is + -- serving.__ Stopping ends the accepting, not the handlers: a + -- caller which has to see them out has to do that itself, before + -- letting the process go. + -- + -- This is installed once per server, and a TCP and a UDP server + -- share nothing but these settings, so running both with one + -- 'ServerSettings' installs it twice. } -- | Default settings. 'settingsOnException' does nothing, so the @@ -218,6 +236,7 @@ { settingsOnException = \_ _ -> return () , settingsGracefulCloseTimeout = 5000 , settingsAcceptRetryDelay = 100000 + , settingsInstallShutdownHandler = const $ return () } -- | Calling 'settingsOnException', never letting it throw. A hook @@ -239,6 +258,43 @@ ---------------------------------------------------------------- +-- | What an accept loop accepts through. +data Listener = Listener + { acceptOrStop :: IO (Maybe (Socket, SockAddr)) + -- ^ Accepts a connection. 'Nothing' instead once the server has + -- been told to stop. + } + +-- | The listener for a socket an accept loop is about to run on. +-- +-- The loop is stopped by being told, through +-- 'settingsInstallShutdownHandler', not by having the socket closed +-- under it. A close ends a thread blocked on a descriptor only because +-- the event manager tells the waiter so, and that is being removed; +-- after it is, a waiter left behind on a number which has been handed to +-- someone else since would accept on that someone else's socket. +-- +-- 'acceptWith' gives up on the accept itself, so nothing has to be +-- closed and nothing has to be interrupted, on any platform. +makeListener :: ServerSettings -> Socket -> IO Listener +makeListener set@ServerSettings{..} sock = do + stopping <- newTVarIO False + settingsInstallShutdownHandler $ atomically $ writeTVar stopping True + return Listener{acceptOrStop = accepting stopping} + where + -- Closing the listening socket is how a server was stopped before + -- there was anything to tell, and callers which do it are still out + -- there. It means what the shutdown action means: stop accepting. + closedAsStopped = E.handle $ \e -> + if isClosedSocketError e then return Nothing else E.throwIO e + accepting stopping = + closedAsStopped $ + either (const Nothing) Just + <$> safeAcceptWith set (acceptWith sock $ told stopping) + told stopping = check =<< readTVar stopping + +---------------------------------------------------------------- + -- | Accepting a connection, retrying on transient errors. -- -- 'accept' fails routinely for reasons which do not mean that the @@ -262,8 +318,7 @@ -- | 'safeAccept' with the accepting action passed explicitly. The -- error paths described above cannot be provoked on a real listening -- socket, so the test suite reaches them through this. -safeAcceptWith - :: ServerSettings -> IO (Socket, SockAddr) -> IO (Socket, SockAddr) +safeAcceptWith :: ServerSettings -> IO a -> IO a safeAcceptWith set@ServerSettings{..} accept' = loop where loop = do @@ -278,12 +333,9 @@ report set Nothing $ E.toException e threadDelay settingsAcceptRetryDelay loop - -- These cost nothing; retry immediately. - | ioeGetErrorType e == Interrupted -> loop - | ioe_errno e == Just connAborted -> loop + -- One connection's failure, which costs nothing to retry. + | isTransientAcceptError e -> loop | otherwise -> E.throwIO e - - Errno connAborted = eCONNABORTED ---------------------------------------------------------------- diff --git a/Network/Run/TCP.hs b/Network/Run/TCP.hs --- a/Network/Run/TCP.hs +++ b/Network/Run/TCP.hs @@ -27,7 +27,7 @@ ) where import qualified Control.Exception as E -import Control.Monad (forever) +import Control.Monad (when) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as NE import Network.Socket @@ -42,7 +42,7 @@ -- created by this function listens on a single address family. Use -- 'runTCPServerWithSocket' with one socket per address to serve both -- IPv4 and IPv6. -runTCPServer :: Maybe HostName -> ServiceName -> (Socket -> IO a) -> IO a +runTCPServer :: Maybe HostName -> ServiceName -> (Socket -> IO a) -> IO () runTCPServer = runTCPServerWithSettings defaultServerSettings -- | Running a TCP server with the given settings. @@ -51,7 +51,7 @@ -> Maybe HostName -> ServiceName -> (Socket -> IO a) - -> IO a + -> IO () runTCPServerWithSettings set mhost port server = do addr <- resolve Stream mhost port [AI_PASSIVE] NE.head E.bracket (openTCPServerSocket addr) close $ \sock -> @@ -63,7 +63,7 @@ -- ^ A listening socket created by 'openTCPServerSocket'. -> (Socket -> IO a) -- ^ Called for each incoming connection, in a new thread - -> IO a + -> IO () runTCPServerWithSocket = runTCPServerWithSocketAndSettings defaultServerSettings -- | Running a TCP server on a given listen socket with the given @@ -74,10 +74,22 @@ -- ^ A listening socket created by 'openTCPServerSocket'. -> (Socket -> IO a) -- ^ Called for each incoming connection, in a new thread - -> IO a -runTCPServerWithSocketAndSettings set sock server = forever $ - E.bracketOnError (safeAccept set sock) (close . fst) $ \(conn, peer) -> - forkConnection set conn peer (labelMe "TCP server" >> server conn) + -> IO () +runTCPServerWithSocketAndSettings set sock server = do + listener <- makeListener set sock + loop listener + where + -- Accepting inside the acquire: once it has returned, the socket + -- is closed if anything goes wrong before the thread owning it has + -- been forked. + loop listener = do + more <- E.bracketOnError (acceptOrStop listener) (mapM_ (close . fst)) $ \maccepted -> + case maccepted of + Nothing -> return False + Just (conn, peer) -> do + forkConnection set conn peer (labelMe "TCP server" >> server conn) + return True + when more $ loop listener ---------------------------------------------------------------- diff --git a/Network/Run/TCP/Timeout.hs b/Network/Run/TCP/Timeout.hs --- a/Network/Run/TCP/Timeout.hs +++ b/Network/Run/TCP/Timeout.hs @@ -16,7 +16,7 @@ ) where import qualified Control.Exception as E -import Control.Monad (forever) +import Control.Monad (when) import qualified Data.List.NonEmpty as NE import Network.Socket import qualified System.TimeManager as T @@ -82,11 +82,21 @@ -- ^ A listening socket created by 'openTCPServerSocket'. -> TimeoutServer () -> IO () -runTCPServerWithSocketAndSettings set tm sock server = - T.withManager (tm * 1000000) $ \mgr -> forever $ - E.bracketOnError (safeAccept set sock) (close . fst) $ \(conn, peer) -> - forkConnection set conn peer (runServer mgr conn) +runTCPServerWithSocketAndSettings set tm sock server = do + listener <- makeListener set sock + T.withManager (tm * 1000000) $ \mgr -> loop listener mgr where + -- Accepting inside the acquire: once it has returned, the socket + -- is closed if anything goes wrong before the thread owning it has + -- been forked. + loop listener mgr = do + more <- E.bracketOnError (acceptOrStop listener) (mapM_ (close . fst)) $ \maccepted -> + case maccepted of + Nothing -> return False + Just (conn, peer) -> do + forkConnection set conn peer (runServer mgr conn) + return True + when more $ loop listener mgr runServer mgr conn = do labelMe "TCP timeout server" T.withHandleKillThread mgr (return ()) $ \th -> server mgr th conn diff --git a/Network/Run/UDP.hs b/Network/Run/UDP.hs --- a/Network/Run/UDP.hs +++ b/Network/Run/UDP.hs @@ -13,8 +13,8 @@ ) where import Control.Concurrent (forkIO) +import Control.Concurrent.STM import qualified Control.Exception as E -import Control.Monad (forever) import Data.ByteString (ByteString) import qualified Data.List.NonEmpty as NE import Network.Socket @@ -60,18 +60,29 @@ -> IO () runUDPServerForkWithSettings _ [] _ _ = return () runUDPServerForkWithSettings set (h : hs) port server = do - mapM_ (forkIO . run) hs - run h + -- One variable for all of the hosts: the shutdown action stops the + -- server, which is all of these loops. + stopping <- newTVarIO False + settingsInstallShutdownHandler set $ atomically $ writeTVar stopping True + mapM_ (forkIO . run stopping) hs + run stopping h where - run host = do + run stopping host = do labelMe $ "UDP server for " ++ host - runUDPServer (Just host) port $ \lsock -> forever $ do - -- An error from 'recvFrom' means that the listening socket - -- itself is gone, so it is left to propagate as before. - (bs0, peeraddr) <- recvFrom lsock 2048 - -- Everything below is per-datagram work. A failure here - -- must not take the entire server down. - dispatch peeraddr bs0 `E.catch` onDispatchError peeraddr + runUDPServer (Just host) port $ \lsock -> loop stopping lsock + + -- The receive and the stop variable are waited for at once, so + -- nothing has to be closed under this thread and nothing has to be + -- thrown into it. + loop stopping lsock = do + r <- recvFromWith lsock 2048 $ check =<< readTVar stopping + case r of + Left () -> return () + Right (bs0, peeraddr) -> do + -- Everything below is per-datagram work. A failure here + -- must not take the entire server down. + dispatch peeraddr bs0 `E.catch` onDispatchError peeraddr + loop stopping lsock onDispatchError peeraddr e = report set (Just peeraddr) $ E.toException (e :: E.IOException) diff --git a/network-run.cabal b/network-run.cabal --- a/network-run.cabal +++ b/network-run.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: network-run -version: 0.6.0 +version: 0.7.0 license: BSD3 license-file: LICENSE maintainer: kazu@iij.ad.jp @@ -31,8 +31,9 @@ build-depends: base >=4 && <5, bytestring >=0.10 && <0.13, - network >=3.2.4 && <3.3, - time-manager >=0.2 && <0.4 + network >=3.2.10 && <3.3, + stm >=2.5 && <2.6, + time-manager >=0.2 && <0.6 executable tcp-client main-is: tcpClient.hs @@ -113,6 +114,7 @@ bytestring, directory, hspec, - network >=3.2.4, - time-manager >=0.2 && <0.4, + network >=3.2.10, + stm >=2.5 && <2.6, + time-manager >=0.2 && <0.6, network-run diff --git a/test/TCPSpec.hs b/test/TCPSpec.hs --- a/test/TCPSpec.hs +++ b/test/TCPSpec.hs @@ -10,7 +10,7 @@ import Data.IORef import Data.List (nub) import qualified Data.List.NonEmpty as NE -import GHC.IO.Exception (IOErrorType (InvalidArgument)) +import GHC.IO.Exception (IOErrorType (InvalidArgument), ioe_errno) import Network.Socket import Network.Socket.ByteString import System.IO.Error (ioeGetErrorType) @@ -76,10 +76,33 @@ close lsock r <- timeout 2000000 $ takeMVar done case r of - Just (Left _) -> return () - Just (Right _) -> expectationFailure "the accept loop returned" + Just (Right _) -> return () + Just (Left e) -> + expectationFailure $ "the accept loop threw: " ++ show e Nothing -> expectationFailure "the accept loop did not stop" + it "stops when the shutdown handler is run" $ limited $ do + told <- newEmptyMVar + done <- newEmptyMVar + let set = + defaultServerSettings + { settingsInstallShutdownHandler = putMVar told + } + withListenSocket $ \lsock _ -> do + void $ + forkFinally + (runTCPServerWithSocketAndSettings set lsock echo) + (putMVar done) + stop <- takeMVar told + threadDelay 100000 + stop + r <- timeout 2000000 $ takeMVar done + case r of + Just (Right _) -> return () + Just (Left e) -> + expectationFailure $ "the accept loop threw: " ++ show e + Nothing -> expectationFailure "the accept loop did not stop" + it "listens on a single address family" $ limited $ do port <- freeTCPPort let hints = @@ -153,8 +176,11 @@ it "sets ReuseAddr" $ limited $ withListenSocket $ \lsock _ -> getSocketOption lsock ReuseAddr `shouldNotReturn` 0 - it "sets close-on-exec" $ limited $ withListenSocket $ \lsock _ -> - withFdSocket lsock getCloseOnExec `shouldReturn` True + it "sets close-on-exec" $ limited $ + if os == "mingw32" + then pendingWith "Windows has no close-on-exec" + else withListenSocket $ \lsock _ -> + withFdSocket lsock getCloseOnExec `shouldReturn` True it "sets a composite option" $ limited $ do addr <- resolve Stream (Just loopback) "0" [AI_PASSIVE] NE.head @@ -270,8 +296,15 @@ ---------------------------------------------------------------- -- | 'accept' on a socket which is not listening fails with @EINVAL@. +-- +-- Winsock numbers its errors apart from @errno@ and network reports them +-- all as 'GHC.IO.Exception.OtherError', so on Windows the code is what +-- there is to look at. invalidArgument :: Selector E.IOException -invalidArgument e = ioeGetErrorType e == InvalidArgument +invalidArgument e = + ioeGetErrorType e == InvalidArgument || ioe_errno e == Just wsaEINVAL + where + wsaEINVAL = 10022 -- | A handler which fails the first connection and echoes the rest. failFirst :: IORef Int -> Socket -> IO () diff --git a/test/TimeoutSpec.hs b/test/TimeoutSpec.hs --- a/test/TimeoutSpec.hs +++ b/test/TimeoutSpec.hs @@ -75,8 +75,9 @@ close lsock r <- timeout 2000000 $ takeMVar done case r of - Just (Left _) -> return () - Just (Right _) -> expectationFailure "the accept loop returned" + Just (Right _) -> return () + Just (Left e) -> + expectationFailure $ "the accept loop threw: " ++ show e Nothing -> expectationFailure "the accept loop did not stop" ---------------------------------------------------------------- diff --git a/test/UDPSpec.hs b/test/UDPSpec.hs --- a/test/UDPSpec.hs +++ b/test/UDPSpec.hs @@ -4,6 +4,7 @@ module UDPSpec (spec) where import Control.Concurrent +import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad import Data.ByteString (ByteString) @@ -25,11 +26,20 @@ it "serves a datagram" $ limited $ do port <- freeUDPPort - let server sock = forever $ do - (bs, peer) <- recvFrom sock 2048 - void $ sendTo sock bs peer - withServerThread (runUDPServer (Just loopback) port server) $ + -- The loop is told to stop rather than killed: on Windows + -- killThread does not reach a thread blocked in recvFrom, + -- and withServerThread would wait for it for ever. + stop <- newTVarIO False + let server sock = loop + where + loop = do + r <- recvFromWith sock 2048 $ check =<< readTVar stop + case r of + Left () -> return () + Right (bs, peer) -> sendTo sock bs peer >> loop + withServerThread (runUDPServer (Just loopback) port server) $ do udpRequest loopback port "hello" `shouldReturn` Just "hello" + atomically $ writeTVar stop True describe "runUDPServerFork" $ do it "serves a datagram" $ @@ -37,6 +47,29 @@ withUDPServerFork defaultServerSettings [loopback] echoDatagram $ \port -> udpRequest loopback port "hello" `shouldReturn` Just "hello" + it "stops when the shutdown handler is run" $ limited $ do + told <- newEmptyMVar + done <- newEmptyMVar + port <- freeUDPPort + let set = + defaultServerSettings + { settingsInstallShutdownHandler = putMVar told + } + void $ + forkFinally + (runUDPServerForkWithSettings set [loopback] port echoDatagram) + (putMVar done) + stop <- takeMVar told + -- Served, so the loop is in its receive, and then told to stop. + udpRequest loopback port "hello" `shouldReturn` Just "hello" + stop + r <- timeout 2000000 $ takeMVar done + case r of + Just (Right ()) -> return () + Just (Left e) -> + expectationFailure $ "the server threw: " ++ show e + Nothing -> expectationFailure "the server did not stop" + it "returns at once when no host is given" $ limited $ do r <- timeout 1000000 $ runUDPServerFork [] "0" $ \_ _ -> return () r `shouldBe` Just () @@ -136,7 +169,11 @@ go 0 _ _ = return Nothing go n sock server = do void $ sendTo sock bs server - mr <- timeout 200000 $ fst <$> recvFrom sock 2048 + -- Not System.Timeout: on Windows an asynchronous exception does + -- not reach a thread blocked in recvFrom, so the retry below + -- would never come. + late <- registerDelay 200000 + mr <- recvFromWith sock 2048 $ check =<< readTVar late case mr of - Just r -> return $ Just r - Nothing -> go (n - 1) sock server + Right (r, _) -> return $ Just r + Left () -> go (n - 1) sock server