e550ddd · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

Merge pull request #185 from yesodweb/breakable-forever

e550dddMichael Snoyman13 years ago

warp/Network/Wai/Handler/Warp/FdCache.hs+21−27

compare with master

unifiedsplit

@@ -23,12 +23,12 @@ module Network.Wai.Handler.Warp.FdCache (

23 ) where

25import Control.Applicative ((<$>), (<*>))

26−import Control.Concurrent (forkIO, threadDelay)

26+import Control.Concurrent (threadDelay)

27−import Control.Exception (bracket)

27+import Control.Exception (bracket, mask_)

28−import Control.Monad (void)

29import Data.Hashable (hash)

28import Data.Hashable (hash)

30import Data.IORef (IORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

29import Data.IORef (IORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

31import Network.Wai.Handler.Warp.MultiMap

30import Network.Wai.Handler.Warp.MultiMap

31+import Network.Wai.Handler.Warp.Thread

32import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)

33import System.Posix.Types (Fd)

@@ -67,17 +67,17 @@ type Hash = Int

67type FdCache = MMap Hash FdEntry

68newtype MutableFdCache = MutableFdCache (IORef FdCache)

70−newMutableFdCache :: IO MutableFdCache

71−newMutableFdCache = MutableFdCache <$> newIORef empty

72−

73fdCache :: MutableFdCache -> IO FdCache

70fdCache :: MutableFdCache -> IO FdCache

74fdCache (MutableFdCache ref) = readIORef ref

71fdCache (MutableFdCache ref) = readIORef ref

76−swapWithNew :: MutableFdCache -> IO FdCache

73+swapWithNew :: IORef FdCache -> IO FdCache

77−swapWithNew (MutableFdCache ref) = atomicModifyIORef ref (\t -> (empty, t))

74+swapWithNew ref = atomicModifyIORef ref (\t -> (empty, t))

79update :: MutableFdCache -> (FdCache -> FdCache) -> IO ()

76update :: 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

81 !_ <- atomicModifyIORef ref $ \t -> let !new = f t in (new, ())

82 return ()

@@ -103,24 +103,16 @@ withFdCache duration action = bracket (initialize duration)

103initialize :: Int -> IO (Maybe MutableFdCache)

104initialize 0 = return Nothing

105initialize duration = do

106− mfc <- newMutableFdCache

106+ ref' <- forkIOwithBreakableForever empty $ \ref -> do

107− -- FIXME: how to stop this thread?

107+ threadDelay duration

108− void . forkIO $ cleanLoop duration mfc

108+ clean ref

109− return (Just mfc)

109+ return (Just (MutableFdCache ref'))

110−

110+

111−cleanLoop :: Int -> MutableFdCache -> IO ()

111+clean :: IORef FdCache -> 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

112+clean ref = do

121− old <- swapWithNew mfc

113+ old <- swapWithNew ref

122 new <- pruneWith old prune

114 new <- pruneWith old prune

123− update mfc (merge new)

115+ update' ref (merge new)

124

116

125prune :: t -> Some FdEntry -> IO [(t, Some FdEntry)]

117prune :: t -> Some FdEntry -> IO [(t, Some FdEntry)]

126prune k v@(One (FdEntry _ fd mst)) = status mst >>= prune'

118prune k v@(One (FdEntry _ fd mst)) = status mst >>= prune'

@@ -142,7 +134,9 @@ prune k (Tom ent@(FdEntry _ fd mst) vs) = status mst >>= prune'

142

134

143terminate :: Maybe MutableFdCache -> IO ()

135terminate :: Maybe MutableFdCache -> IO ()

144terminate Nothing = return ()

136terminate 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

146 where

140 where

147 closeIt (_, FdEntry _ fd _) = closeFd fd

141 closeIt (_, FdEntry _ fd _) = closeFd fd

148

142

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

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

47import GHC.Exts (mkWeak#)

48import GHC.IO (IO (IO))

49#endif

50−import GHC.Weak (Weak (..))

51−import Control.Concurrent (forkIO, threadDelay, myThreadId, killThread)

50+import Control.Concurrent (threadDelay, myThreadId, killThread)

52import qualified Control.Exception as E

51import qualified Control.Exception as E

53−import Control.Monad (forever, void)

54import Data.IORef (IORef)

52import Data.IORef (IORef)

55import qualified Data.IORef as I

53import qualified Data.IORef as I

56−import Data.Typeable (Typeable)

54+import GHC.Weak (Weak (..))

55+import Network.Wai.Handler.Warp.Thread

57import System.Mem.Weak (deRefWeak)

56import System.Mem.Weak (deRefWeak)

59----------------------------------------------------------------

58----------------------------------------------------------------

@@ -74,26 +73,17 @@ data State = Active -- Manager turns it to Inactive.

75----------------------------------------------------------------

74----------------------------------------------------------------

77−data TimeoutManagerStopped = TimeoutManagerStopped

78− deriving (Show, Typeable)

79−instance E.Exception TimeoutManagerStopped

80−

81−----------------------------------------------------------------

82−

83-- | Creating timeout manager which works every N micro seconds

76-- | Creating timeout manager which works every N micro seconds

84-- where N is the first argument.

77-- where N is the first argument.

85initialize :: Int -> IO Manager

78initialize :: Int -> IO Manager

86initialize timeout = do

79initialize timeout = do

87− ref <- I.newIORef []

80+ ref' <- forkIOwithBreakableForever [] $ \ref -> do

88− void . forkIO $ E.handle ignoreStop $ forever $ do

89 threadDelay timeout

81 threadDelay timeout

90 old <- I.atomicModifyIORef ref (\x -> ([], x))

82 old <- I.atomicModifyIORef ref (\x -> ([], x))

91 merge <- prune old id

83 merge <- prune old id

92 I.atomicModifyIORef ref (\new -> (merge new, ()))

84 I.atomicModifyIORef ref (\new -> (merge new, ()))

93− return $ Manager ref

85+ return $ Manager ref'

94 where

86 where

95− ignoreStop TimeoutManagerStopped = return ()

96−

97 prune [] front = return front

87 prune [] front = return front

98 prune (m@(Handle onTimeout iactive):rest) front = do

88 prune (m@(Handle onTimeout iactive):rest) front = do

99 state <- I.atomicModifyIORef iactive (\x -> (inactivate x, x))

89 state <- I.atomicModifyIORef iactive (\x -> (inactivate x, x))

@@ -109,10 +99,8 @@ initialize timeout = do

109----------------------------------------------------------------

99----------------------------------------------------------------

110

100

111stopManager :: Manager -> IO ()

101stopManager :: Manager -> IO ()

112−stopManager (Manager ihandles) = E.mask_ $ do

102+stopManager (Manager ref) = E.mask_ $ do

113− -- Put an undefined value in the IORef to kill the worker thread (yes, it's

103+ !handles <- breakForever ref

114− -- a bit of a hack)

115− !handles <- I.atomicModifyIORef ihandles $ \h -> (E.throw TimeoutManagerStopped, h)

116 mapM_ fire handles

104 mapM_ fire handles

117 where

105 where

118 fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll

106 fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll

warp/test/ThreadSpec.hsnew file+26

@@ -0,0 +1,26 @@

1+{-# LANGUAGE BangPatterns #-}

3+module ThreadSpec where

5+import Control.Concurrent (threadDelay)

6+import Data.IORef (readIORef, writeIORef, atomicModifyIORef)

7+import Network.Wai.Handler.Warp.Thread

8+import Test.Hspec

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

63 Network.Wai.Handler.Warp.Run

64 Network.Wai.Handler.Warp.SendFile

65 Network.Wai.Handler.Warp.Settings

66+ Network.Wai.Handler.Warp.Thread

66 Network.Wai.Handler.Warp.Types

67 Network.Wai.Handler.Warp.Types

67 Paths_warp

68 Paths_warp

68 Ghc-Options: -Wall

69 Ghc-Options: -Wall

@@ -86,6 +87,7 @@ Test-Suite spec

86 RequestSpec

87 RequestSpec

87 ResponseHeaderSpec

88 ResponseHeaderSpec

88 RunSpec

89 RunSpec

90+ ThreadSpec

89 Hs-Source-Dirs: test, .

91 Hs-Source-Dirs: test, .

90 Type: exitcode-stdio-1.0

92 Type: exitcode-stdio-1.0