0e58be3 · typeable/wai

git.iko.soy/typeable/wai

Merge branch for #717

commit0e58be3a0857550eca41a79f4e2d7e71b0a8a9bd authorKazu Yamamoto <kazu@iij.ad.jp> dateFri 26 Apr 2019, 13:16 (7 years ago) parent24332aa 5beeddf

3 files changed+85−24

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+31−11 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+4−4 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+50−9

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+31−11

@@ -109,8 +109,8 @@ frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway

109109 PriorityFrame newpri <- guardIt $ decodePriorityFrame header pl

110110 checkPriority newpri streamId

111111 return True -- just ignore this frame

112− Just strm@Stream{streamState,streamPrecedence} -> do

113− state <- readIORef streamState

112+ Just strm@Stream{streamPrecedence} -> do

113+ state <- readStreamState strm

114114 state' <- stream ftyp header pl ctx state strm

115115 case state' of

116116 Open (NoBody tbl@(_,reqvt) pri) -> do

@@ -119,7 +119,7 @@ frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway

119119 when (just mcl (/= (0 :: Int))) $

120120 E.throwIO $ StreamError ProtocolError streamId

121121 writeIORef streamPrecedence $ toPrecedence pri

122− writeIORef streamState HalfClosed

122+ halfClosedRemote ctx strm

123123 (!req, !ii) <- mkreq tbl (Just 0, return "")

124124 atomically $ writeTQueue inputQ $ Input strm req reqvt ii

125125 Open (HasBody tbl@(_,reqvt) pri) -> do

@@ -128,17 +128,20 @@ frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway

128128 let !mcl = readInt <$> getHeaderValue tokenContentLength reqvt

129129 writeIORef streamPrecedence $ toPrecedence pri

130130 bodyLength <- newIORef 0

131− writeIORef streamState $ Open (Body q mcl bodyLength)

131+ setStreamState ctx strm $ Open (Body q mcl bodyLength)

132132 readQ <- newReadBody q

133133 bodySource <- mkSource readQ

134134 (!req, !ii) <- mkreq tbl (mcl, readSource bodySource)

135135 atomically $ writeTQueue inputQ $ Input strm req reqvt ii

136136 s@(Open Continued{}) -> do

137137 setContinued

138− writeIORef streamState s

139− s -> do -- Idle, Open Body, HalfClosed, Closed

138+ setStreamState ctx strm s

139+ HalfClosedRemote -> do

140140 resetContinued

141− writeIORef streamState s

141+ halfClosedRemote ctx strm

142+ s -> do -- Idle, Open Body, Closed

143+ resetContinued

144+ setStreamState ctx strm s

142145 return True

143146 where

144147 setContinued = writeIORef continued (Just streamId)

@@ -155,8 +158,8 @@ frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway

155158 case mstrm0 of

156159 js@(Just strm0) -> do

157160 when (ftyp == FrameHeaders) $ do

158− st <- readIORef $ streamState strm0

159− when (isHalfClosed st) $ E.throwIO $ ConnectionError StreamClosed "header must not be sent to half closed"

161+ st <- readStreamState strm0

162+ when (isHalfClosedRemote st) $ E.throwIO $ ConnectionError StreamClosed "header must not be sent to half or fully closed stream"

160163 -- Priority made an idele stream

161164 when (isIdle st) $ opened ctx strm0

162165 return js

@@ -286,11 +289,28 @@ stream FrameHeaders header@FrameHeader{flags} bs _ (Open (Body q _ _)) _ = do

286289 let !endOfStream = testEndStream flags

287290 if endOfStream then do

288291 atomically $ writeTQueue q ""

289− return HalfClosed

292+ return HalfClosedRemote

290293 else

291294 -- we don't support continuation here.

292295 E.throwIO $ ConnectionError ProtocolError "continuation in trailer is not supported"

293296

297+-- ignore data-frame except for flow-control when we're done locally

298+stream FrameData

299+ FrameHeader{flags,payloadLength}

300+ _bs

301+ Context{controlQ} s@(HalfClosedLocal _)

302+ Stream{streamNumber} = do

303+ let !endOfStream = testEndStream flags

304+ when (payloadLength /= 0) $ do

305+ let !frame1 = windowUpdateFrame 0 payloadLength

306+ !frame2 = windowUpdateFrame streamNumber payloadLength

307+ !frame = frame1 `BS.append` frame2

308+ enqueueControl controlQ $ CFrame frame

309+ if endOfStream then do

310+ return HalfClosedRemote

311+ else

312+ return s

313+

294314stream FrameData

295315 header@FrameHeader{flags,payloadLength,streamId}

296316 bs

@@ -312,7 +332,7 @@ stream FrameData

312332 Nothing -> return ()

313333 Just cl -> when (cl /= len) $ E.throwIO $ StreamError ProtocolError streamId

314334 atomically $ writeTQueue q ""

315− return HalfClosed

335+ return HalfClosedRemote

316336 else

317337 return s

318338

warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+4−4

@@ -137,7 +137,7 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

137137 off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen

138138 case rspn of

139139 RspnNobody _ _ -> do

140− closed ctx strm Finished

140+ halfClosedLocal ctx strm Finished

141141 return off

142142 RspnFile _ _ path mpart -> do

143143 -- Data frame payload

@@ -174,8 +174,8 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

174174 output _ _ _ = undefined -- never reach

175175

176176 outputOrEnqueueAgain out off = E.handle resetStream $ do

177− state <- readIORef $ streamState strm

178− if isClosed state then

177+ state <- readStreamState strm

178+ if isHalfClosedLocal state then

179179 return off

180180 else case out of

181181 Output _ _ _ wait _ OWait -> do

@@ -292,7 +292,7 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

292292 handleEndOfBody True off0 noTrailers trailers = do

293293 off1 <- handleTrailers noTrailers off0 trailers

294294 void tell

295− closed ctx strm Finished

295+ halfClosedLocal ctx strm Finished

296296 return off1

297297 handleEndOfBody False off0 _ _ = return off0

298298

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+50−9

@@ -166,7 +166,8 @@ data ClosedCode = Finished

166166data StreamState =

167167 Idle

168168 | Open !OpenState

169− | HalfClosed

169+ | HalfClosedRemote

170+ | HalfClosedLocal !ClosedCode

170171 | Closed !ClosedCode

171172 | Reserved

172173

@@ -178,9 +179,15 @@ isOpen :: StreamState -> Bool

178179isOpen Open{} = True

179180isOpen _ = False

180181

181−isHalfClosed :: StreamState -> Bool

182−isHalfClosed HalfClosed = True

183−isHalfClosed _ = False

182+isHalfClosedRemote :: StreamState -> Bool

183+isHalfClosedRemote HalfClosedRemote = True

184+isHalfClosedRemote (Closed _) = True

185+isHalfClosedRemote _ = False

186+

187+isHalfClosedLocal :: StreamState -> Bool

188+isHalfClosedLocal (HalfClosedLocal _) = True

189+isHalfClosedLocal (Closed _) = True

190+isHalfClosedLocal _ = False

184191

185192isClosed :: StreamState -> Bool

186193isClosed Closed{} = True

@@ -189,7 +196,8 @@ isClosed _ = False

189196instance Show StreamState where

190197 show Idle = "Idle"

191198 show Open{} = "Open"

192− show HalfClosed = "HalfClosed"

199+ show HalfClosedRemote = "HalfClosedRemote"

200+ show (HalfClosedLocal e) = "HalfClosedLocal: " ++ show e

193201 show (Closed e) = "Closed: " ++ show e

194202 show Reserved = "Reserved"

195203

@@ -221,16 +229,49 @@ newPushStream Context{serverStreamId} win pre = do

221229

222230----------------------------------------------------------------

223231

232+{-# INLINE readStreamState #-}

233+readStreamState :: Stream -> IO StreamState

234+readStreamState Stream{streamState} =

235+ readIORef streamState

236+

237+{-# INLINE setStreamState #-}

238+setStreamState :: Context -> Stream -> StreamState -> IO ()

239+setStreamState _ Stream{streamState} val = writeIORef streamState val

240+

224241opened :: Context -> Stream -> IO ()

225−opened Context{concurrency} Stream{streamState} = do

242+opened ctx@Context{concurrency} strm = do

226243 atomicModifyIORef' concurrency (\x -> (x+1,()))

227− writeIORef streamState (Open JustOpened)

244+ setStreamState ctx strm (Open JustOpened)

245+

246+halfClosedRemote :: Context -> Stream -> IO ()

247+halfClosedRemote ctx stream@Stream{streamState} = do

248+ !closingCode <- atomicModifyIORef streamState closeHalf

249+ case closingCode of

250+ Nothing -> return ()

251+ Just cc -> closed ctx stream cc

252+ where

253+ closeHalf :: StreamState -> (StreamState, Maybe ClosedCode)

254+ closeHalf x@(Closed _) = (x, Nothing)

255+ closeHalf (HalfClosedLocal cc) = (Closed cc, Just cc)

256+ closeHalf _ = (HalfClosedRemote, Nothing)

257+

258+halfClosedLocal :: Context -> Stream -> ClosedCode -> IO ()

259+halfClosedLocal ctx stream@Stream{streamState} cc = do

260+ shouldFinalize <- atomicModifyIORef streamState closeHalf

261+ when shouldFinalize $

262+ closed ctx stream cc

263+ where

264+ closeHalf :: StreamState -> (StreamState, Bool)

265+ closeHalf x@(Closed _) = (x, False)

266+ closeHalf HalfClosedRemote = (Closed cc, True)

267+ closeHalf _ = (HalfClosedLocal cc, False)

228268

229269closed :: Context -> Stream -> ClosedCode -> IO ()

230−closed Context{concurrency,streamTable} Stream{streamState,streamNumber} cc = do

270+closed ctx@Context{concurrency,streamTable} strm@Stream{streamNumber} cc = do

231271 remove streamTable streamNumber

272+ -- TODO: prevent double-counting

232273 atomicModifyIORef' concurrency (\x -> (x-1,()))

233− writeIORef streamState (Closed cc) -- anyway

274+ setStreamState ctx strm (Closed cc) -- anyway

234275

235276----------------------------------------------------------------

236277