git.iko.soy/typeable/wai
on master help sign in
Merge pull request #185 from yesodweb/breakable-forever

Breakable forever

commite550dddae635c24cd62d98592a790ed5f8cbb720 authorMichael Snoyman <michael@snoyman.com> dateMon 21 Oct 2013, 04:27 (13 years ago) parent9e9cec98184b22
5 files changed+77−46
compare with master
unifiedsplit
warp/Network/Wai/Handler/Warp/FdCache.hs+21−27 warp/Network/Wai/Handler/Warp/Thread.hs+21 warp/Network/Wai/Handler/Warp/Timeout.hs+7−19 warp/test/ThreadSpec.hs+26 warp/warp.cabal+2
warp/Network/Wai/Handler/Warp/FdCache.hs+21−27
@@ -23,12 +23,12 @@ module Network.Wai.Handler.Warp.FdCache (
2323 ) where
2424
2525import Control.Applicative ((<$>), (<*>))
26−import Control.Concurrent (forkIO, threadDelay)
27−import Control.Exception (bracket)
28−import Control.Monad (void)
26+import Control.Concurrent (threadDelay)
27+import Control.Exception (bracket, mask_)
2928import Data.Hashable (hash)
3029import Data.IORef (IORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
3130import Network.Wai.Handler.Warp.MultiMap
31+import Network.Wai.Handler.Warp.Thread
3232import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)
3333import System.Posix.Types (Fd)
3434
@@ -67,17 +67,17 @@ type Hash = Int
6767type FdCache = MMap Hash FdEntry
6868newtype MutableFdCache = MutableFdCache (IORef FdCache)
6969
70−newMutableFdCache :: IO MutableFdCache
71−newMutableFdCache = MutableFdCache <$> newIORef empty
72−
7370fdCache :: MutableFdCache -> IO FdCache
7471fdCache (MutableFdCache ref) = readIORef ref
7572
76−swapWithNew :: MutableFdCache -> IO FdCache
77−swapWithNew (MutableFdCache ref) = atomicModifyIORef ref (\t -> (empty, t))
73+swapWithNew :: IORef FdCache -> IO FdCache
74+swapWithNew ref = atomicModifyIORef ref (\t -> (empty, t))
7875
7976update :: MutableFdCache -> (FdCache -> FdCache) -> IO ()
80−update (MutableFdCache ref) f = do
77+update (MutableFdCache ref) = update' ref
78+
79+update' :: IORef FdCache -> (FdCache -> FdCache) -> IO ()
80+update' ref f = do
8181 !_ <- atomicModifyIORef ref $ \t -> let !new = f t in (new, ())
8282 return ()
8383
@@ -103,24 +103,16 @@ withFdCache duration action = bracket (initialize duration)
103103initialize :: Int -> IO (Maybe MutableFdCache)
104104initialize 0 = return Nothing
105105initialize duration = do
106− mfc <- newMutableFdCache
107− -- FIXME: how to stop this thread?
108− void . forkIO $ cleanLoop duration mfc
109− return (Just mfc)
110−
111−cleanLoop :: Int -> MutableFdCache -> IO ()
112−cleanLoop duration mfc = loop
113− where
114− loop = do
115− clean mfc
116− threadDelay duration
117− loop
118−
119−clean :: MutableFdCache -> IO ()
120−clean mfc = do
121− old <- swapWithNew mfc
106+ ref' <- forkIOwithBreakableForever empty $ \ref -> do
107+ threadDelay duration
108+ clean ref
109+ return (Just (MutableFdCache ref'))
110+
111+clean :: IORef FdCache -> IO ()
112+clean ref = do
113+ old <- swapWithNew ref
122114 new <- pruneWith old prune
123− update mfc (merge new)
115+ update' ref (merge new)
124116
125117prune :: t -> Some FdEntry -> IO [(t, Some FdEntry)]
126118prune k v@(One (FdEntry _ fd mst)) = status mst >>= prune'
@@ -142,7 +134,9 @@ prune k (Tom ent@(FdEntry _ fd mst) vs) = status mst >>= prune'
142134
143135terminate :: Maybe MutableFdCache -> IO ()
144136terminate Nothing = return ()
145−terminate (Just (MutableFdCache mfc)) = readIORef mfc >>= mapM_ closeIt . toList
137+terminate (Just (MutableFdCache ref)) = mask_ $ do
138+ !t <- breakForever ref
139+ mapM_ closeIt $ toList t
146140 where
147141 closeIt (_, FdEntry _ fd _) = closeFd fd
148142
warp/Network/Wai/Handler/Warp/Thread.hsnew file+21
@@ -0,0 +1,21 @@
1+module Network.Wai.Handler.Warp.Thread (
2+ forkIOwithBreakableForever
3+ , breakForever
4+ ) where
5+
6+import Control.Concurrent (forkIO)
7+import Control.Monad (void, forever)
8+import Control.Exception (handle, throw, mask_, SomeException, AsyncException(ThreadKilled))
9+import Data.IORef
10+
11+forkIOwithBreakableForever :: a -> (IORef a -> IO ()) -> IO (IORef a)
12+forkIOwithBreakableForever ini action = do
13+ ref <- newIORef ini
14+ void . forkIO . handle stopPropagation . forever . mask_ $ action ref
15+ return ref
16+
17+stopPropagation :: SomeException -> IO ()
18+stopPropagation _ = return ()
19+
20+breakForever :: IORef a -> IO a
21+breakForever ref = atomicModifyIORef ref $ \x -> (throw ThreadKilled, x)
warp/Network/Wai/Handler/Warp/Timeout.hs+7−19
@@ -47,13 +47,12 @@ import Control.Concurrent (ThreadId(..))
4747import GHC.Exts (mkWeak#)
4848import GHC.IO (IO (IO))
4949#endif
50−import GHC.Weak (Weak (..))
51−import Control.Concurrent (forkIO, threadDelay, myThreadId, killThread)
50+import Control.Concurrent (threadDelay, myThreadId, killThread)
5251import qualified Control.Exception as E
53−import Control.Monad (forever, void)
5452import Data.IORef (IORef)
5553import qualified Data.IORef as I
56−import Data.Typeable (Typeable)
54+import GHC.Weak (Weak (..))
55+import Network.Wai.Handler.Warp.Thread
5756import System.Mem.Weak (deRefWeak)
5857
5958----------------------------------------------------------------
@@ -74,26 +73,17 @@ data State = Active -- Manager turns it to Inactive.
7473
7574----------------------------------------------------------------
7675
77−data TimeoutManagerStopped = TimeoutManagerStopped
78− deriving (Show, Typeable)
79−instance E.Exception TimeoutManagerStopped
80−
81−----------------------------------------------------------------
82−
8376-- | Creating timeout manager which works every N micro seconds
8477-- where N is the first argument.
8578initialize :: Int -> IO Manager
8679initialize timeout = do
87− ref <- I.newIORef []
88− void . forkIO $ E.handle ignoreStop $ forever $ do
80+ ref' <- forkIOwithBreakableForever [] $ \ref -> do
8981 threadDelay timeout
9082 old <- I.atomicModifyIORef ref (\x -> ([], x))
9183 merge <- prune old id
9284 I.atomicModifyIORef ref (\new -> (merge new, ()))
93− return $ Manager ref
85+ return $ Manager ref'
9486 where
95− ignoreStop TimeoutManagerStopped = return ()
96−
9787 prune [] front = return front
9888 prune (m@(Handle onTimeout iactive):rest) front = do
9989 state <- I.atomicModifyIORef iactive (\x -> (inactivate x, x))
@@ -109,10 +99,8 @@ initialize timeout = do
10999----------------------------------------------------------------
110100
111101stopManager :: Manager -> IO ()
112−stopManager (Manager ihandles) = E.mask_ $ do
113− -- Put an undefined value in the IORef to kill the worker thread (yes, it's
114− -- a bit of a hack)
115− !handles <- I.atomicModifyIORef ihandles $ \h -> (E.throw TimeoutManagerStopped, h)
102+stopManager (Manager ref) = E.mask_ $ do
103+ !handles <- breakForever ref
116104 mapM_ fire handles
117105 where
118106 fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll
warp/test/ThreadSpec.hsnew file+26
@@ -0,0 +1,26 @@
1+{-# LANGUAGE BangPatterns #-}
2+
3+module ThreadSpec where
4+
5+import Control.Concurrent (threadDelay)
6+import Data.IORef (readIORef, writeIORef, atomicModifyIORef)
7+import Network.Wai.Handler.Warp.Thread
8+import Test.Hspec
9+
10+main :: IO ()
11+main = hspec spec
12+
13+spec :: Spec
14+spec = describe "forkIOwithBreakableForever" $ do
15+ it "can be breakable" $ do
16+ ref' <- forkIOwithBreakableForever True $ \ref -> do
17+ threadDelay 1000
18+ !_ <- atomicModifyIORef ref (\x -> (False, x))
19+ return ()
20+ threadDelay 100000
21+ readIORef ref' `shouldReturn` False
22+ _ <- breakForever ref'
23+ threadDelay 100000
24+ writeIORef ref' True
25+ threadDelay 100000
26+ readIORef ref' `shouldReturn` True
warp/warp.cabal+2
@@ -63,6 +63,7 @@ Library
6363 Network.Wai.Handler.Warp.Run
6464 Network.Wai.Handler.Warp.SendFile
6565 Network.Wai.Handler.Warp.Settings
66+ Network.Wai.Handler.Warp.Thread
6667 Network.Wai.Handler.Warp.Types
6768 Paths_warp
6869 Ghc-Options: -Wall
@@ -86,6 +87,7 @@ Test-Suite spec
8687 RequestSpec
8788 ResponseHeaderSpec
8889 RunSpec
90+ ThreadSpec
8991 Hs-Source-Dirs: test, .
9092 Type: exitcode-stdio-1.0
9193