on master help sign in
Use Control.Reader for Timeout and FdCache
commitf4fac7cc28db2cf60db3200b48739e2d02af35b2 authorMichael Snoyman <michael@snoyman.com> dateMon 4 Aug 2014, 10:56 (12 years ago) parent8750896
8 files changed+67−116
auto-update/Control/Reaper.hs+16−10 auto-update/test/Control/ReaperSpec.hs+3−1 warp/Network/Wai/Handler/Warp/FdCache.hs+26−28 warp/Network/Wai/Handler/Warp/MultiMap.hs+1−1 warp/Network/Wai/Handler/Warp/Thread.hs−28 warp/Network/Wai/Handler/Warp/Timeout.hs+20−19 warp/test/ThreadSpec.hs−26 warp/warp.cabal+1−3
auto-update/Control/Reaper.hs+16−10
@@ -9,16 +9,16 @@ module Control.Reaper
99 , reaperDelay
1010 , reaperCons
1111 , reaperNull
12+ , reaperEmpty
1213 , mkListAction
1314 ) where
1415
15−import Prelude hiding (null)
16−import Data.Monoid (Monoid, mempty)
16+import Prelude hiding (null, empty)
1717import Control.Monad (join, void)
1818import Data.Function (fix)
1919import Control.AutoUpdate.Util (atomicModifyIORef')
2020import Control.Concurrent (forkIO, threadDelay)
21−import Data.IORef (newIORef)
21+import Data.IORef (IORef, newIORef)
2222import Control.Exception (mask_)
2323import qualified Prelude
2424
@@ -61,6 +61,12 @@ data ReaperSettings workload item = ReaperSettings
6161 -- Default: 'Prelude.null'.
6262 --
6363 -- Since 0.1.1
64+ , reaperEmpty :: workload
65+ -- ^ An empty workload.
66+ --
67+ -- Default: empty list.
68+ --
69+ -- Since 0.1.1
6470 }
6571
6672-- | Default @ReaperSettings@ value, biased towards having a list of work
@@ -73,6 +79,7 @@ defaultReaperSettings = ReaperSettings
7379 , reaperDelay = 30000000
7480 , reaperCons = (:)
7581 , reaperNull = Prelude.null
82+ , reaperEmpty = []
7683 }
7784
7885-- | Create a reaper addition function. This funciton can be used to add
@@ -80,16 +87,15 @@ defaultReaperSettings = ReaperSettings
8087-- for you automatically.
8188--
8289-- Since 0.1.1
83−reaper :: Monoid workload
84− => ReaperSettings workload item
85− -> IO (item -> IO ())
86−reaper (ReaperSettings action delay cons null) = do
90+reaper :: ReaperSettings workload item
91+ -> IO (item -> IO (), IORef (Maybe workload))
92+reaper (ReaperSettings action delay cons null empty) = do
8793 stateRef <- newIORef Nothing
88− return (update stateRef)
94+ return (update stateRef, stateRef)
8995 where
9096 update stateRef item = mask_ $ join $ atomicModifyIORef' stateRef $ \ms ->
9197 case ms of
92− Nothing -> (Just $ cons item mempty, spawn stateRef)
98+ Nothing -> (Just $ cons item empty, spawn stateRef)
9399 Just wl -> (Just $ cons item wl, return ())
94100
95101 spawn stateRef = void $ forkIO $ fix $ \loop -> do
@@ -97,7 +103,7 @@ reaper (ReaperSettings action delay cons null) = do
97103 wl1 <- atomicModifyIORef' stateRef $ \ms ->
98104 case ms of
99105 Nothing -> error "Control.Reaper.reaper.spawn: unexpected Nothing (1)"
100− Just wl -> (Just mempty, wl)
106+ Just wl -> (Just empty, wl)
101107 wl2 <- action wl1
102108 join $ atomicModifyIORef' stateRef $ \ms ->
103109 case ms of
auto-update/test/Control/ReaperSpec.hs+3−1
@@ -10,7 +10,7 @@ type Item = (Int, IORef Int)
1010
1111spec :: Spec
1212spec = prop "works" $ \is -> do
13− addItem <- reaper defaultReaperSettings
13+ (addItem, wlRef) <- reaper defaultReaperSettings
1414 { reaperAction = mkListAction $ \(i, ref) -> do
1515 modifyIORef ref succ
1616 return $ if i > 1
@@ -31,3 +31,5 @@ spec = prop "works" $ \is -> do
3131 actual `shouldBe` (expected :: Int)
3232 threadDelay 100000
3333 mapM_ test tests
34+ Nothing <- readIORef wlRef
35+ return ()
warp/Network/Wai/Handler/Warp/FdCache.hs+26−28
@@ -23,14 +23,14 @@ module Network.Wai.Handler.Warp.FdCache (
2323 ) where
2424
2525import Control.Applicative ((<$>), (<*>))
26−import Control.Concurrent (threadDelay)
27−import Control.Exception (bracket, mask_)
26+import Control.Exception (bracket)
2827import Data.Hashable (hash)
2928import Network.Wai.Handler.Warp.IORef
3029import Network.Wai.Handler.Warp.MultiMap
31−import Network.Wai.Handler.Warp.Thread
3230import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)
3331import System.Posix.Types (Fd)
32+import Control.Reaper
33+import Data.Maybe (fromMaybe)
3434
3535----------------------------------------------------------------
3636
@@ -68,19 +68,12 @@ type Hash = Int
6868type FdCache = MMap Hash FdEntry
6969
7070-- | Mutable Fd cacher.
71−newtype MutableFdCache = MutableFdCache (IORef FdCache)
71+data MutableFdCache = MutableFdCache
72+ ((Hash, FdEntry) -> IO ())
73+ (IORef (Maybe FdCache))
7274
7375fdCache :: MutableFdCache -> IO FdCache
74−fdCache (MutableFdCache ref) = readIORef ref
75−
76−swapWithNew :: IORef FdCache -> IO FdCache
77−swapWithNew ref = atomicModifyIORef' ref $ \t -> (empty, t)
78−
79−update :: MutableFdCache -> (FdCache -> FdCache) -> IO ()
80−update (MutableFdCache ref) = update' ref
81−
82−update' :: IORef FdCache -> (FdCache -> FdCache) -> IO ()
83−update' ref f = atomicModifyIORef' ref $ \t -> (f t, ())
76+fdCache (MutableFdCache _ ref) = fromMaybe empty <$> readIORef ref
8477
8578look :: MutableFdCache -> FilePath -> Hash -> IO (Maybe FdEntry)
8679look mfc path key = searchWith key check <$> fdCache mfc
@@ -106,17 +99,19 @@ withFdCache duration action = bracket (initialize duration)
10699-- The first argument is a cache duration in second.
107100initialize :: Int -> IO (Maybe MutableFdCache)
108101initialize 0 = return Nothing
109−initialize duration = do
110− ref' <- forkIOwithBreakableForever empty $ \ref -> do
111− threadDelay duration
112− clean ref
113− return (Just (MutableFdCache ref'))
114−
115−clean :: IORef FdCache -> IO ()
116−clean ref = do
117− old <- swapWithNew ref
102+initialize duration = fmap (Just . uncurry MutableFdCache)
103+ $ reaper defaultReaperSettings
104+ { reaperAction = clean
105+ , reaperDelay = duration
106+ , reaperCons = uncurry insert
107+ , reaperNull = isEmpty
108+ , reaperEmpty = empty
109+ }
110+
111+clean :: FdCache -> IO (FdCache -> FdCache)
112+clean old = do
118113 new <- pruneWith old prune
119− update' ref (merge new)
114+ return $ merge new
120115
121116prune :: t -> Some FdEntry -> IO [(t, Some FdEntry)]
122117prune k v@(One (FdEntry _ fd mst)) = status mst >>= prune'
@@ -138,8 +133,11 @@ prune k (Tom ent@(FdEntry _ fd mst) vs) = status mst >>= prune'
138133
139134terminate :: Maybe MutableFdCache -> IO ()
140135terminate Nothing = return ()
141−terminate (Just (MutableFdCache ref)) = mask_ $ do
142− !t <- breakForever ref
136+terminate (Just (MutableFdCache _ ref)) = do
137+ !t <- atomicModifyIORef ref $ \mm ->
138+ case mm of
139+ Nothing -> (Nothing, empty)
140+ Just m -> (Just empty, m)
143141 mapM_ closeIt $ toList t
144142 where
145143 closeIt (_, FdEntry _ fd _) = closeFd fd
@@ -148,12 +146,12 @@ terminate (Just (MutableFdCache ref)) = mask_ $ do
148146
149147-- | Getting 'Fd' and 'Refresh' from the mutable Fd cacher.
150148getFd :: MutableFdCache -> FilePath -> IO (Fd, Refresh)
151−getFd mfc path = look mfc path key >>= getFd'
149+getFd mfc@(MutableFdCache add _) path = look mfc path key >>= getFd'
152150 where
153151 key = hash path
154152 getFd' Nothing = do
155153 ent@(FdEntry _ fd mst) <- newFdEntry path
156− update mfc (insert key ent)
154+ add (key, ent)
157155 return (fd, refresh mst)
158156 getFd' (Just (FdEntry _ fd mst)) = do
159157 refresh mst
warp/Network/Wai/Handler/Warp/MultiMap.hs+1−1
@@ -72,7 +72,7 @@ searchWith xk f (Node _ l k v r) = case compare xk k of
7272----------------------------------------------------------------
7373
7474-- | O(1)
75−isEmpty :: (Eq k, Eq v) => MMap k v -> Bool
75+isEmpty :: MMap k v -> Bool
7676isEmpty Leaf = True
7777isEmpty _ = False
7878
warp/Network/Wai/Handler/Warp/Thread.hsdeleted−28
@@ -1,28 +0,0 @@
1−{-# LANGUAGE DeriveDataTypeable #-}
2−
3−module Network.Wai.Handler.Warp.Thread (
4− forkIOwithBreakableForever
5− , breakForever
6− ) where
7−
8−import Control.Concurrent (forkIO)
9−import Control.Exception (handle, throw, mask_, Exception)
10−import Control.Monad (void, forever)
11−import Data.IORef
12−import Data.Typeable
13−
14−data BreakForever = BreakForever deriving (Show, Typeable)
15−
16−instance Exception BreakForever
17−
18−forkIOwithBreakableForever :: a -> (IORef a -> IO ()) -> IO (IORef a)
19−forkIOwithBreakableForever ini action = do
20− ref <- newIORef ini
21− void . forkIO . handle stopPropagation . forever . mask_ $ action ref
22− return ref
23−
24−stopPropagation :: BreakForever -> IO ()
25−stopPropagation _ = return ()
26−
27−breakForever :: IORef a -> IO a
28−breakForever ref = atomicModifyIORef ref $ \x -> (throw BreakForever, x)
warp/Network/Wai/Handler/Warp/Timeout.hs+20−19
@@ -49,19 +49,19 @@ import GHC.Conc (ThreadId(..))
4949import GHC.Exts (mkWeak#)
5050import GHC.IO (IO (IO))
5151#endif
52−import Control.Concurrent (threadDelay, myThreadId)
52+import Control.Concurrent (myThreadId)
5353import qualified Control.Exception as E
5454import GHC.Weak (Weak (..))
5555import Network.Wai.Handler.Warp.IORef (IORef)
5656import qualified Network.Wai.Handler.Warp.IORef as I
57−import Network.Wai.Handler.Warp.Thread
5857import System.Mem.Weak (deRefWeak)
5958import Data.Typeable (Typeable)
59+import Control.Reaper
6060
6161----------------------------------------------------------------
6262
6363-- | A timeout manager
64−newtype Manager = Manager (IORef [Handle])
64+data Manager = Manager (Handle -> IO ()) (IORef (Maybe [Handle]))
6565
6666-- | An action to be performed on timeout.
6767type TimeoutAction = IO ()
@@ -79,23 +79,21 @@ data State = Active -- Manager turns it to Inactive.
7979-- | Creating timeout manager which works every N micro seconds
8080-- where N is the first argument.
8181initialize :: Int -> IO Manager
82−initialize timeout = do
83− ref' <- forkIOwithBreakableForever [] $ \ref -> do
84− threadDelay timeout
85− old <- I.atomicModifyIORef' ref (\x -> ([], x))
86− merge <- prune old id
87− I.atomicModifyIORef' ref (\new -> (merge new, ()))
88− return $ Manager ref'
82+initialize timeout =
83+ fmap (uncurry Manager) $ reaper defaultReaperSettings
84+ { reaperAction = mkListAction prune
85+ , reaperDelay = timeout
86+ }
8987 where
90− prune [] front = return front
91− prune (m@(Handle onTimeout iactive):rest) front = do
88+ prune m@(Handle onTimeout iactive) = do
9289 state <- I.atomicModifyIORef' iactive (\x -> (inactivate x, x))
9390 case state of
9491 Inactive -> do
9592 onTimeout `E.catch` ignoreAll
96− prune rest front
97− Canceled -> prune rest front
98− _ -> prune rest (front . (:) m)
93+ return Nothing
94+ Canceled -> return Nothing
95+ _ -> return $ Just m
96+
9997 inactivate Active = Inactive
10098 inactivate x = x
10199
@@ -103,8 +101,11 @@ initialize timeout = do
103101
104102-- | Stopping timeout manager.
105103stopManager :: Manager -> IO ()
106−stopManager (Manager ref) = E.mask_ $ do
107− !handles <- breakForever ref
104+stopManager (Manager _ ref) = E.mask_ $ do
105+ handles <- I.atomicModifyIORef ref $ \x ->
106+ case x of
107+ Nothing -> (Nothing, [])
108+ Just y -> (Just [], y)
108109 mapM_ fire handles
109110 where
110111 fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll
@@ -116,10 +117,10 @@ ignoreAll _ = return ()
116117
117118-- | Registering a timeout action.
118119register :: Manager -> TimeoutAction -> IO Handle
119−register (Manager ref) onTimeout = do
120+register (Manager add _) onTimeout = do
120121 iactive <- I.newIORef Active
121122 let h = Handle onTimeout iactive
122− I.atomicModifyIORef' ref (\x -> (h : x, ()))
123+ add h
123124 return h
124125
125126-- | Registering a timeout action of killing this thread.
warp/test/ThreadSpec.hsdeleted−26
@@ -1,26 +0,0 @@
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+1−3
@@ -40,7 +40,7 @@ Flag warp-debug
4040Library
4141 Build-Depends: base >= 3 && < 5
4242 , array
43− , auto-update >= 0.1 && < 0.2
43+ , auto-update >= 0.1.1 && < 0.2
4444 , blaze-builder >= 0.3.3 && < 0.4
4545 , bytestring >= 0.9.1.4
4646 , case-insensitive >= 0.2
@@ -76,7 +76,6 @@ Library
7676 Network.Wai.Handler.Warp.Run
7777 Network.Wai.Handler.Warp.SendFile
7878 Network.Wai.Handler.Warp.Settings
79− Network.Wai.Handler.Warp.Thread
8079 Network.Wai.Handler.Warp.Types
8180 Network.Wai.Handler.Warp.Windows
8281 Paths_warp
@@ -115,7 +114,6 @@ Test-Suite spec
115114 ResponseHeaderSpec
116115 ResponseSpec
117116 RunSpec
118− ThreadSpec
119117 Hs-Source-Dirs: test, .
120118 Type: exitcode-stdio-1.0
121119