@@ -1,8 +1,12 @@
11{-# LANGUAGE DeriveDataTypeable #-}
22{-# LANGUAGE RecordWildCards, NamedFieldPuns #-}
3−{-# LANGUAGE PatternGuards #-}
3+{-# LANGUAGE PatternGuards, BangPatterns #-}
44
5−module Network.Wai.Handler.Warp.HTTP2.Worker where
5+module Network.Wai.Handler.Warp.HTTP2.Worker (
6+ Responder
7+ , response
8+ , worker
9+ ) where
610
711import Control.Concurrent
812import Control.Concurrent.STM
@@ -15,6 +19,7 @@ import Network.HTTP2
1519import Network.HTTP2.Priority
1620import Network.Wai
1721import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
22+import Network.Wai.Handler.Warp.HTTP2.Manager
1823import Network.Wai.Handler.Warp.HTTP2.Types
1924import Network.Wai.Handler.Warp.IORef
2025import qualified Network.Wai.Handler.Warp.Response as R
@@ -24,16 +29,21 @@ import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceive
2429
2530----------------------------------------------------------------
2631
27−type Responder = T.Handle -> Stream -> Priority -> Request -> Response -> IO ResponseReceived
32+type Responder = ThreadContinue -> T.Handle -> Stream -> Priority -> Request -> Response -> IO ResponseReceived
2833
29−response :: Context -> Responder
30−response Context{outputQ} th strm pri req rsp = do
34+response :: Context -> Manager -> Responder
35+response Context{outputQ} mgr tconf th strm pri req rsp = do
3136 case rsp of
3237 ResponseStream _ _ strmbdy -> do
33− -- fixme: spawn a new worker thread.
34− --
3538 -- We must not exit this WAI application.
3639 -- If the application exits, streaming would be also closed.
40+ -- So, this work occupies this thread.
41+ --
42+ -- We need to increase the number of workers.
43+ myThreadId >>= replace mgr
44+ -- After this work, this thread stops to decease
45+ -- the number of workers.
46+ setThreadContinue tconf False
3747 -- Since 'StreamingBody' is loop, we cannot control it.
3848 -- So, let's serialize 'Builder' with a designated queue.
3949 sq <- newTBQueueIO 10 -- fixme: hard coding: 10
@@ -50,6 +60,7 @@ response Context{outputQ} th strm pri req rsp = do
5060 strmbdy push flush
5161 atomically $ writeTBQueue sq SFinish
5262 _ -> do
63+ setThreadContinue tconf True
5364 let hasBody = requestMethod req /= H.methodHead
5465 || R.hasBody (responseStatus rsp)
5566 enqueue outputQ (OResponse strm rsp (Oneshot hasBody)) pri
@@ -62,34 +73,37 @@ instance Exception Break
6273worker :: Context -> S.Settings -> T.Manager -> Application -> Responder -> IO ()
6374worker ctx@Context{inputQ,outputQ} set tm app responder = do
6475 tid <- myThreadId
65− ref <- newIORef Nothing
76+ sinfo <- newStreamInfo
77+ tcont <- newThreadContinue
6678 let setup = T.register tm $ E.throwTo tid Break
67− E.bracket setup T.cancel $ go ref
79+ E.bracket setup T.cancel $ go sinfo tcont
6880 where
69− go ref th = do
81+ go sinfo tcont th = do
82+ setThreadContinue tcont True
7083 ex <- E.try $ do
7184 T.pause th
7285 Input strm req pri <- atomically $ readTQueue inputQ
73− writeIORef ref $ Just (strm,req)
86+ setStreamInfo sinfo strm req
7487 T.resume th
7588 T.tickle th
76− app req $ responder th strm pri req
77− cont <- case ex of
89+ app req $ responder tcont th strm pri req
90+ cont1 <- case ex of
7891 Right ResponseReceived -> return True
7992 Left e@(SomeException _)
8093 | Just Break <- E.fromException e -> do
81− cleanup ref Nothing
94+ cleanup sinfo Nothing
8295 return True
8396 -- killed by the sender
8497 | Just ThreadKilled <- E.fromException e -> do
85− cleanup ref Nothing
98+ cleanup sinfo Nothing
8699 return False
87100 | otherwise -> do
88− cleanup ref (Just e)
101+ cleanup sinfo (Just e)
89102 return True
90− when cont $ go ref th
91− cleanup ref me = do
92− m <- readIORef ref
103+ cont2 <- getThreadContinue tcont
104+ when (cont1 && cont2) $ go sinfo tcont th
105+ cleanup sinfo me = do
106+ m <- getStreamInfo sinfo
93107 case m of
94108 Nothing -> return ()
95109 Just (strm,req) -> do
@@ -99,7 +113,7 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
99113 case me of
100114 Nothing -> return ()
101115 Just e -> S.settingsOnException set (Just req) e
102− writeIORef ref Nothing
116+ clearStreamInfo sinfo
103117
104118waiter :: TVar Sync -> TBQueue Sequence
105119 -> (Output -> Priority -> IO ()) -> Stream -> Priority
@@ -122,3 +136,31 @@ waiter tvar sq enq strm pri = do
122136 enq (ONext strm next) pri
123137 waiter tvar sq enq strm pri
124138
139+----------------------------------------------------------------
140+
141+newtype ThreadContinue = ThreadContinue (IORef Bool)
142+
143+newThreadContinue :: IO ThreadContinue
144+newThreadContinue = ThreadContinue <$> newIORef True
145+
146+setThreadContinue :: ThreadContinue -> Bool -> IO ()
147+setThreadContinue (ThreadContinue ref) x = writeIORef ref x
148+
149+getThreadContinue :: ThreadContinue -> IO Bool
150+getThreadContinue (ThreadContinue ref) = readIORef ref
151+
152+----------------------------------------------------------------
153+
154+newtype StreamInfo = StreamInfo (IORef (Maybe (Stream,Request)))
155+
156+newStreamInfo :: IO StreamInfo
157+newStreamInfo = StreamInfo <$> newIORef Nothing
158+
159+clearStreamInfo :: StreamInfo -> IO ()
160+clearStreamInfo (StreamInfo ref) = writeIORef ref Nothing
161+
162+setStreamInfo :: StreamInfo -> Stream -> Request -> IO ()
163+setStreamInfo (StreamInfo ref) strm req = writeIORef ref $ Just (strm,req)
164+
165+getStreamInfo :: StreamInfo -> IO (Maybe (Stream, Request))
166+getStreamInfo (StreamInfo ref) = readIORef ref