@@ -5,7 +5,7 @@
5{-# LANGUAGE CPP #-}
5{-# LANGUAGE CPP #-}
7module Network.Wai.Handler.Warp.HTTP2.Worker (
7module Network.Wai.Handler.Warp.HTTP2.Worker (
@@ -19,6 +19,7 @@ import Control.Exception (Exception, SomeException(..), AsyncException(..))
19import qualified Control.Exception as E
19import qualified Control.Exception as E
20import Control.Monad (void, when)
20import Control.Monad (void, when)
21import Data.Typeable
21import Data.Typeable
22+import qualified Network.HTTP.Types as H
22import Network.HTTP2
23import Network.HTTP2
23import Network.HTTP2.Priority
24import Network.HTTP2.Priority
24import Network.Wai.HTTP2 (Http2Application, Response, absurd)
25import Network.Wai.HTTP2 (Http2Application, Response, absurd)
@@ -33,28 +34,36 @@ import qualified Network.Wai.Handler.Warp.Timeout as T
33----------------------------------------------------------------
34----------------------------------------------------------------
35-- | An 'Http2Application' takes a @forall a. Response a -> IO a@; this type
36-- | An 'Http2Application' takes a @forall a. Response a -> IO a@; this type
36−-- implements that with extra arguments.
37+-- implements that by currying some internal arguments.
37−type Responder = forall a. ThreadContinue -> T.Handle -> Stream ->
38+--
39+-- This is the argument to a 'Responder'.
40+type Respond = IO () -> Stream -> TBQueue Sequence ->
38− TBQueue Sequence -> Response a -> IO a
41+ (forall a. Response a -> IO a)
40-- | This function is passed to workers.
43-- | This function is passed to workers.
41-- They also pass 'Response's from 'Http2Application's to this function.
44-- They also pass 'Response's from 'Http2Application's to this function.
42-- This function enqueues commands for the HTTP/2 sender.
45-- This function enqueues commands for the HTTP/2 sender.
43−response :: Context -> Manager -> Responder
46+response :: Context -> Manager -> ThreadContinue -> Respond
44−response Context{outputQ} mgr tconf th strm sq (s, h, strmbdy) = do
47+response ctx mgr tconf tickle strm sq rsp = do
45− -- TODO(awpr)r HEAD requests will still stream.
48+ -- TODO(awpr) HEAD requests will still stream.
47 -- We must not exit this WAI application.
50 -- We must not exit this WAI application.
48 -- If the application exits, streaming would be also closed.
51 -- If the application exits, streaming would be also closed.
49 -- So, this work occupies this thread.
52 -- So, this work occupies this thread.
51 -- We need to increase the number of workers.
54 -- We need to increase the number of workers.
52 myThreadId >>= replaceWithAction mgr
55 myThreadId >>= replaceWithAction mgr
53− -- After this work, this thread stops to decease
56+ -- After this work, this thread stops to decrease the number of workers.
54− -- the number of workers.
55 setThreadContinue tconf False
57 setThreadContinue tconf False
59+ runStream ctx OResponse tickle strm sq rsp
62+ -> (Stream -> H.Status -> H.ResponseHeaders -> Aux -> Output)
64+runStream Context{outputQ} mkOutput tickle strm sq (s, h, strmbdy) = do
56 tvar <- newTVarIO SyncNone
65 tvar <- newTVarIO SyncNone
57− let out = OResponse strm s h (Persist sq tvar)
66+ let out = mkOutput strm s h (Persist sq tvar)
58 -- Since we must not enqueue an empty queue to the priority
67 -- Since we must not enqueue an empty queue to the priority
59 -- queue, we spawn a thread to ensure that the designated
68 -- queue, we spawn a thread to ensure that the designated
60 -- queue is not empty.
69 -- queue is not empty.
@@ -62,7 +71,7 @@ response Context{outputQ} mgr tconf th strm sq (s, h, strmbdy) = do
62 atomically $ writeTVar tvar (SyncNext out)
71 atomically $ writeTVar tvar (SyncNext out)
63 let push b = do
72 let push b = do
64 atomically $ writeTBQueue sq (SBuilder b)
73 atomically $ writeTBQueue sq (SBuilder b)
65− T.tickle th
74+ tickle
66 flush = atomically $ writeTBQueue sq SFlush
75 flush = atomically $ writeTBQueue sq SFlush
67 strmbdy push flush
76 strmbdy push flush
@@ -83,9 +92,9 @@ worker :: Context
83 -> S.Settings
92 -> S.Settings
84 -> T.Manager
93 -> T.Manager
85 -> Http2Application
94 -> Http2Application
86− -> Responder
95+ -> (ThreadContinue -> Respond)
88−worker ctx@Context{inputQ,outputQ} set tm app responder = do
97+worker ctx@Context{inputQ} set tm app respond = do
89 tid <- myThreadId
98 tid <- myThreadId
90 sinfo <- newStreamInfo
99 sinfo <- newStreamInfo
91 tcont <- newThreadContinue
100 tcont <- newThreadContinue
@@ -103,7 +112,7 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
103 setStreamInfo sinfo strm req
112 setStreamInfo sinfo strm req
104 T.resume th
113 T.resume th
105 T.tickle th
114 T.tickle th
106− app (req, flip (const absurd)) $ responder tcont th strm sq
115+ app (req, flip (const absurd)) $ respond tcont (T.tickle th) strm sq
107 cont1 <- case ex of
116 cont1 <- case ex of
108 Right trailers -> do
117 Right trailers -> do
109 atomically $ writeTBQueue sq $ SFinish trailers
118 atomically $ writeTBQueue sq $ SFinish trailers
@@ -156,7 +165,7 @@ waiter tvar sq strm outQ = do
156-- | It would nice if responders could return values to workers.
165-- | It would nice if responders could return values to workers.
157-- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0.
166-- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0.
158-- It is not wise to change this type.
167-- It is not wise to change this type.
159−-- So, a reference is shared by a responder and its worker.
168+-- So, a reference is shared by a 'Respond' and its worker.
160-- The reference refers a value of this type as a return value.
169-- The reference refers a value of this type as a return value.
161-- If 'True', the worker continue to serve requests.
170-- If 'True', the worker continue to serve requests.
162-- Otherwise, the worker get finished.
171-- Otherwise, the worker get finished.