git.iko.soy/typeable/wai
on master help sign in
Wait for pushed stream to be sent before returning
64486f8Andrew Pritchard
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+12−9
compare with master
unifiedsplit
@@ -7,9 +7,10 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
7#if __GLASGOW_HASKELL__ < 709
7#if __GLASGOW_HASKELL__ < 709
8import Control.Applicative
8import Control.Applicative
9#endif
9#endif
10+import Control.Concurrent.MVar (putMVar)
10import Control.Concurrent.STM
11import Control.Concurrent.STM
11import qualified Control.Exception as E
12import qualified Control.Exception as E
12−import Control.Monad (unless, void, when)
13+import Control.Monad (void, when)
13import qualified Data.ByteString as BS
14import qualified Data.ByteString as BS
14import qualified Data.ByteString.Builder as B (int32BE)
15import qualified Data.ByteString.Builder as B (int32BE)
15import qualified Data.ByteString.Builder.Extra as B
16import qualified Data.ByteString.Builder.Extra as B
@@ -49,17 +50,18 @@ type OpenFile = Fd
49
50
50----------------------------------------------------------------
51----------------------------------------------------------------
51
52
52−unlessClosed :: Connection -> Stream -> IO () -> IO ()
53+unlessClosed :: Connection -> Stream -> IO () -> IO Bool
53unlessClosed Connection{connSendAll}
54unlessClosed Connection{connSendAll}
54 strm@Stream{streamState,streamNumber}
55 strm@Stream{streamState,streamNumber}
55 body = E.handle resetStream $ do
56 body = E.handle resetStream $ do
56 state <- readIORef streamState
57 state <- readIORef streamState
57− unless (isClosed state) body
58+ if (isClosed state) then return False else body >> return True
58 where
59 where
59 resetStream e = do
60 resetStream e = do
60 closed strm (ResetByMe e)
61 closed strm (ResetByMe e)
61 let rst = resetFrame InternalError streamNumber
62 let rst = resetFrame InternalError streamNumber
62 connSendAll rst
63 connSendAll rst
64+ return False
63
65
64getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
66getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
65getWindowSize connWindow strmWindow = do
67getWindowSize connWindow strmWindow = do
@@ -98,20 +100,20 @@ frameSender ctx@Context{outputQ,connectionWindow}
98 connSendAll frame
100 connSendAll frame
99 loop
101 loop
100 switch (OResponse strm s h aux) = do
102 switch (OResponse strm s h aux) = do
101− unlessClosed conn strm $
103+ _ <- unlessClosed conn strm $
102 getWindowSize connectionWindow (streamWindow strm) >>=
104 getWindowSize connectionWindow (streamWindow strm) >>=
103 sendResponse strm s h aux
105 sendResponse strm s h aux
104 loop
106 loop
105 switch (ONext strm curr) = do
107 switch (ONext strm curr) = do
106− unlessClosed conn strm $ do
108+ _ <- unlessClosed conn strm $ do
107 lim <- getWindowSize connectionWindow (streamWindow strm)
109 lim <- getWindowSize connectionWindow (streamWindow strm)
108 -- Data frame payload
110 -- Data frame payload
109 Next datPayloadLen mnext <- curr lim
111 Next datPayloadLen mnext <- curr lim
110 fillDataHeaderSend strm 0 datPayloadLen mnext
112 fillDataHeaderSend strm 0 datPayloadLen mnext
111 maybeEnqueueNext strm mnext
113 maybeEnqueueNext strm mnext
112 loop
114 loop
113− switch (OPush oldStrm push strm s h aux) = do
115+ switch (OPush oldStrm push mvar strm s h aux) = do
114− unlessClosed conn oldStrm $ do
116+ pushed <- unlessClosed conn oldStrm $ do
115 lim <- getWindowSize connectionWindow (streamWindow strm)
117 lim <- getWindowSize connectionWindow (streamWindow strm)
116 -- Write and send the promise.
118 -- Write and send the promise.
117 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push
119 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push
@@ -120,9 +122,10 @@ frameSender ctx@Context{outputQ,connectionWindow}
120 -- TODO(awpr): refactor sendResponse to be able to handle non-zero
122 -- TODO(awpr): refactor sendResponse to be able to handle non-zero
121 -- initial offsets and use that to potentially avoid the extra syscall.
123 -- initial offsets and use that to potentially avoid the extra syscall.
122 sendResponse strm s h aux lim
124 sendResponse strm s h aux lim
125+ putMVar mvar pushed
123 loop
126 loop
124 switch (OTrailers strm []) = do
127 switch (OTrailers strm []) = do
125− unlessClosed conn strm $ do
128+ _ <- unlessClosed conn strm $ do
126 -- An empty data frame should be used to end the stream instead
129 -- An empty data frame should be used to end the stream instead
127 -- when there are no trailers.
130 -- when there are no trailers.
128 let flag = setEndStream defaultFlags
131 let flag = setEndStream defaultFlags
@@ -131,7 +134,7 @@ frameSender ctx@Context{outputQ,connectionWindow}
131 flushN frameHeaderLength
134 flushN frameHeaderLength
132 loop
135 loop
133 switch (OTrailers strm trailers) = do
136 switch (OTrailers strm trailers) = do
134− unlessClosed conn strm $ do
137+ _ <- unlessClosed conn strm $ do
135 -- Trailers always indicate the end of a stream; send them in
138 -- Trailers always indicate the end of a stream; send them in
136 -- consecutive header+continuation frames and end the stream.
139 -- consecutive header+continuation frames and end the stream.
137 builder <- hpackEncodeCIHeaders ctx trailers
140 builder <- hpackEncodeCIHeaders ctx trailers
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+9−7
@@ -8,6 +8,7 @@ import Data.ByteString.Builder (Builder)
8import Control.Applicative ((<$>),(<*>))
8import Control.Applicative ((<$>),(<*>))
9#endif
9#endif
10import Control.Concurrent (forkIO)
10import Control.Concurrent (forkIO)
11+import Control.Concurrent.MVar (MVar)
11import Control.Concurrent.STM
12import Control.Concurrent.STM
12import Control.Exception (SomeException)
13import Control.Exception (SomeException)
13import Control.Monad (void)
14import Control.Monad (void)
@@ -69,19 +70,20 @@ data Output = OFinish
69 | OResponse Stream H.Status H.ResponseHeaders Aux
70 | OResponse Stream H.Status H.ResponseHeaders Aux
70 -- ^ Send the headers and as much of the response as is immediately
71 -- ^ Send the headers and as much of the response as is immediately
71 -- available.
72 -- available.
72− | OPush Stream PushPromise Stream H.Status H.ResponseHeaders Aux
73+ | OPush Stream PushPromise (MVar Bool) Stream H.Status H.ResponseHeaders Aux
73− -- ^ Send a PUSH_PROMISE frame, then act like OResponse.
74+ -- ^ Send a PUSH_PROMISE frame, then act like OResponse; signal the
75+ -- MVar whether the promise has been sent.
74 | ONext Stream DynaNext
76 | ONext Stream DynaNext
75 -- ^ Send a chunk of the response.
77 -- ^ Send a chunk of the response.
76 | OTrailers Stream Trailers
78 | OTrailers Stream Trailers
77 -- ^ Send a series of trailers in header and continuation frames.
79 -- ^ Send a series of trailers in header and continuation frames.
78
80
79outputStream :: Output -> Stream
81outputStream :: Output -> Stream
80−outputStream (OResponse strm _ _ _) = strm
82+outputStream (OResponse strm _ _ _) = strm
81−outputStream (ONext strm _) = strm
83+outputStream (ONext strm _) = strm
82−outputStream (OPush strm _ _ _ _ _) = strm
84+outputStream (OPush strm _ _ _ _ _ _) = strm
83−outputStream (OTrailers strm _) = strm
85+outputStream (OTrailers strm _) = strm
84−outputStream _ = error "outputStream"
86+outputStream _ = error "outputStream"
85
87
86----------------------------------------------------------------
88----------------------------------------------------------------
87
89
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+44−21
@@ -93,37 +93,60 @@ cleanupStream Context{outputQ} set strm req me = do
93 Nothing -> return ()
93 Nothing -> return ()
94 Just e -> S.settingsOnException set req e
94 Just e -> S.settingsOnException set req e
95
95
96+-- | Push the given 'Responder' to the client if the settings allow it
97+-- (specifically 'enablePush' and 'maxConcurrentStreams'). Returns 'True' if
98+-- the stream was actually pushed.
99+--
100+-- This is the push function given to an 'HTTP2Application'.
96pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
101pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
97pushResponder ctx set strm promise responder = do
102pushResponder ctx set strm promise responder = do
98 let Context{ http2settings
103 let Context{ http2settings
99− , nextPushStreamId
100 , pushConcurrency
104 , pushConcurrency
101− , streamTable
102 } = ctx
105 } = ctx
103 cnt <- readIORef pushConcurrency
106 cnt <- readIORef pushConcurrency
104 settings <- readIORef http2settings
107 settings <- readIORef http2settings
105 let enabled = enablePush settings
108 let enabled = enablePush settings
106 fits = maybe True (cnt <) $ maxConcurrentStreams settings
109 fits = maybe True (cnt <) $ maxConcurrentStreams settings
107 canPush = fits && enabled
110 canPush = fits && enabled
108− when canPush $ do
111+ if canPush then
112+ actuallyPushResponder ctx set strm promise responder
113+ else
114+ return False
115+
116+-- | Set up a pushed stream and run the 'Responder' in its own thread. Waits
117+-- for the sender thread to handle the push request. This can fail to push the
118+-- stream and return 'False' if the sender dequeued the push request after the
119+-- associated stream was closed.
120+actuallyPushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
121+actuallyPushResponder ctx set strm promise responder = do
122+ let Context{ http2settings
123+ , nextPushStreamId
124+ , pushConcurrency
125+ , streamTable
126+ } = ctx
109− -- Claim the next outgoing stream.
127+ -- Claim the next outgoing stream.
110− newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)
128+ newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)
111− ws <- initialWindowSize <$> readIORef http2settings
129+ ws <- initialWindowSize <$> readIORef http2settings
112−
130+
113− newStrm <- newStream pushConcurrency newSid ws
131+ newStrm <- newStream pushConcurrency newSid ws
114− writeIORef (streamPriority newStrm) $
132+ writeIORef (streamPriority newStrm) $
115− Priority False (streamNumber strm) 16
133+ Priority False (streamNumber strm) 16
116− opened newStrm
134+ opened newStrm
117− insert streamTable newSid newStrm
135+ insert streamTable newSid newStrm
118−
136+
137+ -- Set up a channel for the sender to report back whether it pushed the
138+ -- stream.
139+ mvar <- newEmptyMVar
140+
119− let mkOutput = OPush strm promise
141+ let mkOutput = OPush strm promise mvar
120− tickle = return ()
142+ tickle = return ()
121− respond = runStream ctx mkOutput
143+ respond = runStream ctx mkOutput
122−
144+
123− -- TODO(awpr): synthesize a Request for 'settingsOnException'?
145+ -- TODO(awpr): synthesize a Request for 'settingsOnException'?
124− runResponder responder respond tickle newStrm `E.catch`
146+ _ <- forkIO $ runResponder responder respond tickle newStrm `E.catch`
125− (cleanupStream ctx set strm Nothing . Just)
147+ (cleanupStream ctx set strm Nothing . Just)
126− return canPush
148+
149+ takeMVar mvar
127
150
128data Break = Break deriving (Show, Typeable)
151data Break = Break deriving (Show, Typeable)
129
152