473457b · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

refactoring Output.

473457bKazu Yamamoto10 years ago

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

compare with master

unifiedsplit

@@ -122,16 +122,16 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

122 Nothing -> return ()

123 Just siz -> setLimitForEncoding siz encodeDynamicTable

124

125− output (ONext strm curr mtbq tell) off0 lim = do

125+ output out@(Output strm _ _ tell (ONext curr)) off0 lim = do

126 -- Data frame payload

127 let !buf = connWriteBuffer `plusPtr` off0

128 !siz = connBufferSize - off0

129 Next datPayloadLen mnext <- curr buf siz lim

130 off <- fillDataHeader strm off0 datPayloadLen mnext tell

131− maybeEnqueueNext strm mtbq mnext tell

131+ maybeEnqueueNext out mnext

132 return off

133

134− output (ORspn strm rspn ii tell) off0 lim = do

134+ output out@(Output strm rspn ii tell ORspn) off0 lim = do

135 -- Header frame and Continuation frame

136 let !sid = streamNumber strm

137 !endOfStream = case rspn of

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

150 Next datPayloadLen mnext <-

151 fillFileBodyGetNext conn ii payloadOff lim path mpart

152 off' <- fillDataHeader strm off datPayloadLen mnext tell

153− maybeEnqueueNext strm Nothing mnext tell

153+ maybeEnqueueNext out mnext

154 return off'

155 RspnBuilder _ _ builder -> do

156 -- Data frame payload

@@ -158,23 +158,23 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

158 Next datPayloadLen mnext <-

159 fillBuilderBodyGetNext conn ii payloadOff lim builder

160 off' <- fillDataHeader strm off datPayloadLen mnext tell

161− maybeEnqueueNext strm Nothing mnext tell

161+ maybeEnqueueNext out mnext

162 return off'

163 RspnStreaming _ _ tbq -> do

164 let payloadOff = off + frameHeaderLength

165 Next datPayloadLen mnext <-

166 fillStreamBodyGetNext conn payloadOff lim tbq strm

167 off' <- fillDataHeader strm off datPayloadLen mnext tell

168− maybeEnqueueNext strm (Just tbq) mnext tell

168+ maybeEnqueueNext out mnext

169 return off'

170

171− output (OPush strm ths rspn ii tell pid) off0 lim = do

171+ output (Output strm rspn ii tell (OPush ths pid)) off0 lim = do

172 -- Creating a push promise header

173 -- Frame id should be associated stream id from the client.

174 let !sid = streamNumber strm

175 len <- pushPromise pid sid ths off0

176 off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + len

177− output (ORspn strm rspn ii tell) off lim

177+ output (Output strm rspn ii tell ORspn) off lim

178

179 output _ _ _ = undefined -- never reach

180

@@ -183,9 +183,9 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

183 if isClosed state then

184 return off

185 else case out of

186− OWait strm' rsp ii wait -> do

186+ Output strm' rsp ii wait OWait -> do

187 -- Checking if all push are done.

188− let out' = ORspn strm' rsp ii (return ())

188+ let out' = Output strm' rsp ii (return ()) ORspn

189 forkAndEnqueueWhenReady wait outputQ out' mgr

190 return off

191 _ -> case mtbq of

@@ -252,12 +252,11 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}

252

253 {-# INLINE maybeEnqueueNext #-}

254 -- Re-enqueue the stream in the output queue.

255− maybeEnqueueNext :: Stream -> Maybe (TBQueue Sequence)

256− -> Maybe DynaNext -> IO () -> IO ()

255+ maybeEnqueueNext :: Output -> Maybe DynaNext -> IO ()

257− maybeEnqueueNext _ _ Nothing _ = return ()

256+ maybeEnqueueNext _ Nothing = return ()

258− maybeEnqueueNext strm mtbq (Just next) tell = enqueueOutput outputQ out

257+ maybeEnqueueNext out (Just next) = enqueueOutput outputQ out'

259 where

258 where

260− !out = ONext strm next mtbq tell

259+ !out' = out { outputType = ONext next }

261

260

262 {-# INLINE sendHeadersIfNecessary #-}

261 {-# INLINE sendHeadersIfNecessary #-}

263 -- Send headers if there is not room for a 1-byte data frame, and return

262 -- Send headers if there is not room for a 1-byte data frame, and return

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+14−19

@@ -68,27 +68,22 @@ rspnHeaders (RspnStreaming _ t _) = t

68rspnHeaders (RspnBuilder _ t _) = t

69rspnHeaders (RspnFile _ t _ _ ) = t

71−data Output = ORspn !Stream !Rspn !InternalInfo (IO ()) -- done

71+data Output = Output {

72− | OWait !Stream !Rspn !InternalInfo (IO ()) -- done

72+ outputStream :: !Stream

73− | OPush !Stream -- stream for this push from this server

73+ , outputRspn :: !Rspn

74− TokenHeaderList

75− !Rspn {- RspnFile only-}

76− !InternalInfo (IO ()) -- wait for done

74+ , outputII :: !InternalInfo

75+ , outputHook :: IO () -- OPush: wait for done, O*: telling done

76+ , outputType :: !OutputType

77+ }

78+

79+data OutputType = ORspn

80+ | OWait

77− !StreamId -- associated stream id from client

81+ | OPush !TokenHeaderList !StreamId -- associated stream id from client

78− | ONext !Stream !DynaNext !(Maybe (TBQueue Sequence)) (IO ()) -- done

82+ | ONext !DynaNext

79−

80−outputStream :: Output -> Stream

81−outputStream (ORspn strm _ _ _) = strm

82−outputStream (OPush strm _ _ _ _ _) = strm

83−outputStream (OWait strm _ _ _) = strm

84−outputStream (ONext strm _ _ _) = strm

86outputMaybeTBQueue :: Output -> Maybe (TBQueue Sequence)

84outputMaybeTBQueue :: Output -> Maybe (TBQueue Sequence)

87−outputMaybeTBQueue (ORspn _ (RspnStreaming _ _ tbq) _ _) = Just tbq

85+outputMaybeTBQueue (Output _ (RspnStreaming _ _ tbq) _ _ _) = Just tbq

88−outputMaybeTBQueue (ORspn _ _ _ _) = Nothing

86+outputMaybeTBQueue _ = Nothing

89−outputMaybeTBQueue (OPush _ _ _ _ _ _) = Nothing

90−outputMaybeTBQueue (OWait _ _ _ _) = Nothing

91−outputMaybeTBQueue (ONext _ _ mtbq _) = mtbq

93data Control = CFinish

88data Control = CFinish

94 | CGoaway !ByteString

89 | CGoaway !ByteString

warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+12−11

@@ -57,7 +57,7 @@ type Responder = InternalInfo

57pushStream :: Context -> S.Settings

58 -> StreamId -> ValueTable -> Request -> InternalInfo

59 -> Maybe HTTP2Data

60− -> IO (Stream -> Rspn -> InternalInfo -> IO () -> Output, IO ())

60+ -> IO (OutputType, IO ())

61pushStream _ _ _ _ _ _ Nothing = return (ORspn, return ())

62pushStream ctx@Context{http2settings,outputQ,streamTable}

63 settings pid reqvt req ii (Just h2d)

@@ -110,7 +110,8 @@ pushStream ctx@Context{http2settings,outputQ,streamTable}

110 !ths = (tokenLastModified,date) :

111 addContentHeadersForFilePart ths0 part

112 pushLogger req path size

113+ let !ot = OPush promisedRequest pid

113− let out = OPush strm promisedRequest rsp ii (increment tvar) pid

114+ !out = Output strm rsp ii (increment tvar) ot

114 enqueueOutput outputQ out

115 enqueueOutput outputQ out

115 push tvar pps (n + 1)

116 push tvar pps (n + 1)

116

117

@@ -156,8 +157,8 @@ response settings ctx@Context{outputQ} mgr ii reqvt tconf strm req rsp = case rs

156 responseNoBody' s tbl = do

157 responseNoBody' s tbl = do

157 logger req s Nothing

158 logger req s Nothing

158 setThreadContinue tconf True

159 setThreadContinue tconf True

159− let rspn = RspnNobody s tbl

160+ let !rspn = RspnNobody s tbl

160− out = ORspn strm rspn ii (return ())

161+ !out = Output strm rspn ii (return ()) ORspn

161 enqueueOutput outputQ out

162 enqueueOutput outputQ out

162 return ResponseReceived

163 return ResponseReceived

163

164

@@ -165,8 +166,8 @@ response settings ctx@Context{outputQ} mgr ii reqvt tconf strm req rsp = case rs

165 logger req s Nothing

166 logger req s Nothing

166 setThreadContinue tconf True

167 setThreadContinue tconf True

167 tbl <- toHeaderTable hs0

168 tbl <- toHeaderTable hs0

168− let rspn = RspnBuilder s tbl bdy

169+ let !rspn = RspnBuilder s tbl bdy

169− out = rspnOrWait strm rspn ii tell

170+ !out = Output strm rspn ii tell rspnOrWait

170 enqueueOutput outputQ out

171 enqueueOutput outputQ out

171 return ResponseReceived

172 return ResponseReceived

172

173

@@ -191,8 +192,8 @@ response settings ctx@Context{outputQ} mgr ii reqvt tconf strm req rsp = case rs

191 | otherwise = do

192 | otherwise = do

192 logger req s (filePartByteCount <$> mpart)

193 logger req s (filePartByteCount <$> mpart)

193 setThreadContinue tconf True

194 setThreadContinue tconf True

194− let rspn = RspnFile s tbl path mpart

195+ let !rspn = RspnFile s tbl path mpart

195− out = rspnOrWait strm rspn ii tell

196+ !out = Output strm rspn ii tell rspnOrWait

196 enqueueOutput outputQ out

197 enqueueOutput outputQ out

197 return ResponseReceived

198 return ResponseReceived

198

199

@@ -217,8 +218,8 @@ response settings ctx@Context{outputQ} mgr ii reqvt tconf strm req rsp = case rs

217 -- So, let's serialize 'Builder' with a designated queue.

218 -- So, let's serialize 'Builder' with a designated queue.

218 tbq <- newTBQueueIO 10 -- fixme: hard coding: 10

219 tbq <- newTBQueueIO 10 -- fixme: hard coding: 10

219 tbl <- toHeaderTable hs0

220 tbl <- toHeaderTable hs0

220− let rspn = RspnStreaming s0 tbl tbq

221+ let !rspn = RspnStreaming s0 tbl tbq

221− out = rspnOrWait strm rspn ii tell

222+ !out = Output strm rspn ii tell rspnOrWait

222 enqueueOutput outputQ out

223 enqueueOutput outputQ out

223 let push b = do

224 let push b = do

224 atomically $ writeTBQueue tbq (SBuilder b)

225 atomically $ writeTBQueue tbq (SBuilder b)

@@ -265,7 +266,7 @@ worker ctx@Context{inputQ,controlQ} set app responder tm = do

265 Nothing -> return ()

266 Nothing -> return ()

266 Just (Input strm req _reqvt _ii) -> do

267 Just (Input strm req _reqvt _ii) -> do

267 closed ctx strm Killed

268 closed ctx strm Killed

268− let frame = resetFrame InternalError (streamNumber strm)

269+ let !frame = resetFrame InternalError (streamNumber strm)

269 enqueueControl controlQ $ CFrame frame

270 enqueueControl controlQ $ CFrame frame

270 case me of

271 case me of

271 Nothing -> return ()

272 Nothing -> return ()