git.iko.soy/typeable/wai
on master help sign in
Wait for pushed stream to be sent before returning

This means we can no longer fall victim to the following race: - Enqueue an OPush on the priority queue and return from pushResponder. - Send the entirety of the (very short) body and close the stream. - Dequeue the OPush and ignore it because the stream is closed. Now if you call pushResponder before finishing the stream, it will have been decided whether the push will succeed before control returns from pushResponder. Of course, the push can still fail if it would exceed the max concurrency, if pushes are disabled, if the peer resets the associated stream, or if you call pushResponder after finishing the response.

commit64486f82ba7c279504bd2fa5773ccd74476e8943 authorAndrew Pritchard <awpr@google.com> dateSat 29 Aug 2015, 13:30 (11 years ago) parent8a9b8f9
3 files changed+65−37
compare with master
unifiedsplit
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+12−9 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+9−7 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+44−21
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+12−9
@@ -7,9 +7,10 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
77#if __GLASGOW_HASKELL__ < 709
88import Control.Applicative
99#endif
10+import Control.Concurrent.MVar (putMVar)
1011import Control.Concurrent.STM
1112import qualified Control.Exception as E
12−import Control.Monad (unless, void, when)
13+import Control.Monad (void, when)
1314import qualified Data.ByteString as BS
1415import qualified Data.ByteString.Builder as B (int32BE)
1516import qualified Data.ByteString.Builder.Extra as B
@@ -49,17 +50,18 @@ type OpenFile = Fd
4950
5051----------------------------------------------------------------
5152
52−unlessClosed :: Connection -> Stream -> IO () -> IO ()
53+unlessClosed :: Connection -> Stream -> IO () -> IO Bool
5354unlessClosed Connection{connSendAll}
5455 strm@Stream{streamState,streamNumber}
5556 body = E.handle resetStream $ do
5657 state <- readIORef streamState
57− unless (isClosed state) body
58+ if (isClosed state) then return False else body >> return True
5859 where
5960 resetStream e = do
6061 closed strm (ResetByMe e)
6162 let rst = resetFrame InternalError streamNumber
6263 connSendAll rst
64+ return False
6365
6466getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
6567getWindowSize connWindow strmWindow = do
@@ -98,20 +100,20 @@ frameSender ctx@Context{outputQ,connectionWindow}
98100 connSendAll frame
99101 loop
100102 switch (OResponse strm s h aux) = do
101− unlessClosed conn strm $
103+ _ <- unlessClosed conn strm $
102104 getWindowSize connectionWindow (streamWindow strm) >>=
103105 sendResponse strm s h aux
104106 loop
105107 switch (ONext strm curr) = do
106− unlessClosed conn strm $ do
108+ _ <- unlessClosed conn strm $ do
107109 lim <- getWindowSize connectionWindow (streamWindow strm)
108110 -- Data frame payload
109111 Next datPayloadLen mnext <- curr lim
110112 fillDataHeaderSend strm 0 datPayloadLen mnext
111113 maybeEnqueueNext strm mnext
112114 loop
113− switch (OPush oldStrm push strm s h aux) = do
114− unlessClosed conn oldStrm $ do
115+ switch (OPush oldStrm push mvar strm s h aux) = do
116+ pushed <- unlessClosed conn oldStrm $ do
115117 lim <- getWindowSize connectionWindow (streamWindow strm)
116118 -- Write and send the promise.
117119 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push
@@ -120,9 +122,10 @@ frameSender ctx@Context{outputQ,connectionWindow}
120122 -- TODO(awpr): refactor sendResponse to be able to handle non-zero
121123 -- initial offsets and use that to potentially avoid the extra syscall.
122124 sendResponse strm s h aux lim
125+ putMVar mvar pushed
123126 loop
124127 switch (OTrailers strm []) = do
125− unlessClosed conn strm $ do
128+ _ <- unlessClosed conn strm $ do
126129 -- An empty data frame should be used to end the stream instead
127130 -- when there are no trailers.
128131 let flag = setEndStream defaultFlags
@@ -131,7 +134,7 @@ frameSender ctx@Context{outputQ,connectionWindow}
131134 flushN frameHeaderLength
132135 loop
133136 switch (OTrailers strm trailers) = do
134− unlessClosed conn strm $ do
137+ _ <- unlessClosed conn strm $ do
135138 -- Trailers always indicate the end of a stream; send them in
136139 -- consecutive header+continuation frames and end the stream.
137140 builder <- hpackEncodeCIHeaders ctx trailers
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+9−7
@@ -8,6 +8,7 @@ import Data.ByteString.Builder (Builder)
88import Control.Applicative ((<$>),(<*>))
99#endif
1010import Control.Concurrent (forkIO)
11+import Control.Concurrent.MVar (MVar)
1112import Control.Concurrent.STM
1213import Control.Exception (SomeException)
1314import Control.Monad (void)
@@ -69,19 +70,20 @@ data Output = OFinish
6970 | OResponse Stream H.Status H.ResponseHeaders Aux
7071 -- ^ Send the headers and as much of the response as is immediately
7172 -- available.
72− | OPush Stream PushPromise Stream H.Status H.ResponseHeaders Aux
73− -- ^ Send a PUSH_PROMISE frame, then act like OResponse.
73+ | OPush Stream PushPromise (MVar Bool) Stream H.Status H.ResponseHeaders Aux
74+ -- ^ Send a PUSH_PROMISE frame, then act like OResponse; signal the
75+ -- MVar whether the promise has been sent.
7476 | ONext Stream DynaNext
7577 -- ^ Send a chunk of the response.
7678 | OTrailers Stream Trailers
7779 -- ^ Send a series of trailers in header and continuation frames.
7880
7981outputStream :: Output -> Stream
80−outputStream (OResponse strm _ _ _) = strm
81−outputStream (ONext strm _) = strm
82−outputStream (OPush strm _ _ _ _ _) = strm
83−outputStream (OTrailers strm _) = strm
84−outputStream _ = error "outputStream"
82+outputStream (OResponse strm _ _ _) = strm
83+outputStream (ONext strm _) = strm
84+outputStream (OPush strm _ _ _ _ _ _) = strm
85+outputStream (OTrailers strm _) = strm
86+outputStream _ = error "outputStream"
8587
8688----------------------------------------------------------------
8789
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+44−21
@@ -93,37 +93,60 @@ cleanupStream Context{outputQ} set strm req me = do
9393 Nothing -> return ()
9494 Just e -> S.settingsOnException set req e
9595
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'.
96101pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
97102pushResponder ctx set strm promise responder = do
98103 let Context{ http2settings
99− , nextPushStreamId
100104 , pushConcurrency
101− , streamTable
102105 } = ctx
103106 cnt <- readIORef pushConcurrency
104107 settings <- readIORef http2settings
105108 let enabled = enablePush settings
106109 fits = maybe True (cnt <) $ maxConcurrentStreams settings
107110 canPush = fits && enabled
108− when canPush $ do
109− -- Claim the next outgoing stream.
110− newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)
111− ws <- initialWindowSize <$> readIORef http2settings
112−
113− newStrm <- newStream pushConcurrency newSid ws
114− writeIORef (streamPriority newStrm) $
115− Priority False (streamNumber strm) 16
116− opened newStrm
117− insert streamTable newSid newStrm
118−
119− let mkOutput = OPush strm promise
120− tickle = return ()
121− respond = runStream ctx mkOutput
122−
123− -- TODO(awpr): synthesize a Request for 'settingsOnException'?
124− runResponder responder respond tickle newStrm `E.catch`
125− (cleanupStream ctx set strm Nothing . Just)
126− return canPush
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
127+ -- Claim the next outgoing stream.
128+ newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)
129+ ws <- initialWindowSize <$> readIORef http2settings
130+
131+ newStrm <- newStream pushConcurrency newSid ws
132+ writeIORef (streamPriority newStrm) $
133+ Priority False (streamNumber strm) 16
134+ opened newStrm
135+ insert streamTable newSid newStrm
136+
137+ -- Set up a channel for the sender to report back whether it pushed the
138+ -- stream.
139+ mvar <- newEmptyMVar
140+
141+ let mkOutput = OPush strm promise mvar
142+ tickle = return ()
143+ respond = runStream ctx mkOutput
144+
145+ -- TODO(awpr): synthesize a Request for 'settingsOnException'?
146+ _ <- forkIO $ runResponder responder respond tickle newStrm `E.catch`
147+ (cleanupStream ctx set strm Nothing . Just)
148+
149+ takeMVar mvar
127150
128151data Break = Break deriving (Show, Typeable)
129152