network-run 0.6.0 → 0.7.0

raw patch · 9 files changed

+232/−53 lines, 9 filesdep +stmdep ~networkPVP ok

version bump matches the API change (PVP)

Dependencies added: stm

Dependency ranges changed: network

API changes (from Hackage documentation)

Files

• CHANGELOG.md +21/−0
• Network/Run/Core.hs +61/−9
• Network/Run/TCP.hs +20/−8
• Network/Run/TCP/Timeout.hs +15/−5
• Network/Run/UDP.hs +22/−11
• network-run.cabal +7/−5
• test/TCPSpec.hs +39/−6
• test/TimeoutSpec.hs +3/−2
• test/UDPSpec.hs +44/−7

@@ -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

@@ -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  ---------------------------------------------------------------- 

@@ -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  ---------------------------------------------------------------- 

@@ -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

@@ -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)

@@ -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

@@ -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 ()

@@ -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"  ----------------------------------------------------------------

@@ -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