git.iko.soy/typeable/wai
on master help sign in
enqueuing response of streams whose window size is greater than 0.

INVARIANT: streams in the output queue have non-zero window size. Old: dequeue, check window, forkIO, wait and then enqueue again. New: check window, forkIO, wait and then enqueue.

commita3eb0562b8772d7e9fabb400cfcc702cb57c480a authorKazu Yamamoto <kazu@iij.ad.jp> dateTue 11 Aug 2015, 22:05 (11 years ago) parent4e0731d
3 files changed+75−55
compare with master
unifiedsplit
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+35−36 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+17−1 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+23−18
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+35−36
@@ -58,22 +58,15 @@ unlessClosed ctx
5858 let rst = resetFrame InternalError streamNumber
5959 connSendAll rst
6060
61−checkWindowSize :: TVar WindowSize -> TVar WindowSize -> PriorityTree Output -> Output -> Priority -> (WindowSize -> IO ()) -> IO ()
62−checkWindowSize connWindow strmWindow outQ out pri body = do
61+checkWindowSize :: TVar WindowSize -> TVar WindowSize -> (WindowSize -> IO ()) -> IO ()
62+checkWindowSize connWindow strmWindow body = do
6363 cw <- atomically $ do
6464 w <- readTVar connWindow
6565 check (w > 0)
6666 return w
67+ -- sw is greater than 0
6768 sw <- atomically $ readTVar strmWindow
68− if sw == 0 then
69− void $ forkIO $ do
70− -- checkme: if connection is closed? GC throws dead lock error?
71− atomically $ do
72− x <- readTVar strmWindow
73− check (x > 0)
74− enqueue outQ out pri
75− else
76− body (min cw sw)
69+ body (min cw sw)
7770
7871frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
7972frameSender ctx@Context{outputQ,connectionWindow}
@@ -100,9 +93,8 @@ frameSender ctx@Context{outputQ,connectionWindow}
10093 switch (OFrame frame) = do
10194 connSendAll frame
10295 loop
103− switch out@(OResponse strm rsp aux) = unlessClosed ctx conn strm $ do
104− pri <- readIORef $ streamPriority strm
105− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
96+ switch (OResponse strm rsp aux) = unlessClosed ctx conn strm $ do
97+ checkWindowSize connectionWindow (streamWindow strm) $ \lim -> do
10698 -- Header frame and Continuation frame
10799 let sid = streamNumber strm
108100 endOfStream = case aux of
@@ -118,7 +110,7 @@ frameSender ctx@Context{outputQ,connectionWindow}
118110 Next datPayloadLen mnext <-
119111 fillResponseBodyGetNext conn ii payloadOff lim rsp
120112 fillDataHeaderSend strm total datPayloadLen mnext
121− maybeEnqueueNext strm mnext pri
113+ maybeEnqueueNext strm mnext
122114 Oneshot False -> do
123115 -- "closed" must be before "connSendAll". If not,
124116 -- the context would be switched to the receiver,
@@ -129,22 +121,21 @@ frameSender ctx@Context{outputQ,connectionWindow}
129121 (off, needSend) <- sendHeadersIfNecessary total
130122 let payloadOff = off + frameHeaderLength
131123 Next datPayloadLen mnext <-
132− fillStreamBodyGetNext conn payloadOff lim sq tvar
124+ fillStreamBodyGetNext conn payloadOff lim sq tvar strm
133125 -- If no data was immediately available, avoid sending an
134126 -- empty data frame.
135127 if datPayloadLen > 0 then
136128 fillDataHeaderSend strm total datPayloadLen mnext
137129 else
138130 when needSend $ flushN off
139− maybeEnqueueNext strm mnext pri
131+ maybeEnqueueNext strm mnext
140132 loop
141− switch out@(ONext strm curr) = unlessClosed ctx conn strm $ do
142− pri <- readIORef $ streamPriority strm
143− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
133+ switch (ONext strm curr) = unlessClosed ctx conn strm $ do
134+ checkWindowSize connectionWindow (streamWindow strm) $ \lim -> do
144135 -- Data frame payload
145136 Next datPayloadLen mnext <- curr lim
146137 fillDataHeaderSend strm 0 datPayloadLen mnext
147− maybeEnqueueNext strm mnext pri
138+ maybeEnqueueNext strm mnext
148139 loop
149140
150141 -- Flush the connection buffer to the socket, where the first 'n' bytes of
@@ -189,9 +180,16 @@ frameSender ctx@Context{outputQ,connectionWindow}
189180 -- available; do nothing otherwise. If the stream is not finished, it must
190181 -- already have been written to the 'TVar' owned by 'waiter', which will
191182 -- put it back into the queue when more output becomes available.
192− maybeEnqueueNext :: Stream -> Control DynaNext -> Priority -> IO ()
193− maybeEnqueueNext strm (CNext next) = enqueue outputQ (ONext strm next)
194− maybeEnqueueNext _ _ = const $ return ()
183+ maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
184+ maybeEnqueueNext strm (CNext next) = do
185+ let out = ONext strm next
186+ sw <- atomically $ readTVar $ streamWindow strm
187+ if sw == 0 then
188+ void $ forkIO $ enqueueWhenWindowIsOpen outputQ out
189+ else do
190+ pri <- readIORef $ streamPriority strm
191+ enqueue outputQ out pri
192+ maybeEnqueueNext _ _ = return ()
195193
196194
197195 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -285,13 +283,13 @@ fileStartEnd _ (Just part) =
285283
286284----------------------------------------------------------------
287285
288−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> IO Next
286+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
289287fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
290− off lim sq tvar = do
288+ off lim sq tvar strm = do
291289 let datBuf = connWriteBuffer `plusPtr` off
292290 room = min (connBufferSize - off) lim
293291 (leftover, cont, len) <- runStreamBuilder datBuf room sq
294− nextForStream connWriteBuffer connBufferSize sq tvar leftover cont len
292+ nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
295293
296294----------------------------------------------------------------
297295
@@ -345,8 +343,8 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0
345343 Just SFlush -> return (LZero, True, total)
346344 Just SFinish -> return (LZero, False, total)
347345
348−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> DynaNext
349−fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
346+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
347+fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
350348 let payloadBuf = buf0 `plusPtr` frameHeaderLength
351349 room0 = min (siz0 - frameHeaderLength) lim0
352350 case leftover0 of
@@ -364,7 +362,7 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
364362 void $ copy payloadBuf bs1
365363 getNext (LTwo bs2 writer) True room0
366364 where
367− getNext = nextForStream buf0 siz0 sq tvar
365+ getNext = nextForStream buf0 siz0 sq tvar strm
368366 write writer1 buf room sofar = do
369367 (len, signal) <- writer1 buf room
370368 case signal of
@@ -379,17 +377,18 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
379377 let !total = sofar + len
380378 getNext (LTwo bs writer) True total
381379
382−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync
380+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
383381 -> Leftover -> Bool -> BytesFilled
384382 -> IO Next
385−nextForStream _ _ _ tvar _ False len = do
383+nextForStream _ _ _ tvar _ _ False len = do
386384 atomically $ writeTVar tvar SyncFinish
387385 return $ Next len CFinish
388−nextForStream buf siz sq tvar LZero True len = do
389− atomically $ writeTVar tvar $ SyncNext (fillBufStream buf siz LZero sq tvar)
386+nextForStream buf siz sq tvar strm LZero True len = do
387+ let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
388+ atomically $ writeTVar tvar $ SyncNext out
390389 return $ Next len CNone
391−nextForStream buf siz sq tvar leftover True len =
392− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar))
390+nextForStream buf siz sq tvar strm leftover True len =
391+ return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
393392
394393----------------------------------------------------------------
395394
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+17−1
@@ -64,6 +64,11 @@ data Output = OFinish
6464 | OResponse Stream Response Aux
6565 | ONext Stream DynaNext
6666
67+outputStream :: Output -> Stream
68+outputStream (OResponse strm _ _) = strm
69+outputStream (ONext strm _) = strm
70+outputStream _ = error "outputStream"
71+
6772----------------------------------------------------------------
6873
6974data Sequence = SFinish
@@ -72,7 +77,7 @@ data Sequence = SFinish
7277
7378data Sync = SyncNone
7479 | SyncFinish
75− | SyncNext DynaNext
80+ | SyncNext Output
7681
7782data Aux = Oneshot Bool
7883 | Persist (TBQueue Sequence) (TVar Sync)
@@ -229,3 +234,14 @@ insert strmtbl k v = reaperAdd strmtbl (k,v)
229234
230235search :: StreamTable -> M.Key -> IO (Maybe Stream)
231236search strmtbl k = M.lookup k <$> reaperRead strmtbl
237+
238+
239+-- INVARIANT: streams in the output queue have non-zero window size.
240+enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()
241+enqueueWhenWindowIsOpen outQ out = do
242+ let strm = outputStream out
243+ atomically $ do
244+ x <- readTVar $ streamWindow strm
245+ check (x > 0)
246+ pri <- readIORef $ streamPriority strm
247+ enqueue outQ out pri
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+23−18
@@ -59,24 +59,29 @@ response Context{outputQ} mgr tconf th strm req rsp = do
5959 -- So, let's serialize 'Builder' with a designated queue.
6060 sq <- newTBQueueIO 10 -- fixme: hard coding: 10
6161 tvar <- newTVarIO SyncNone
62− pri <- readIORef $ streamPriority strm
63− enqueue outputQ (OResponse strm rsp (Persist sq tvar)) pri
62+ let out = OResponse strm rsp (Persist sq tvar)
63+ -- Since we must not enqueue an empty queue to the priority
64+ -- queue, we spawn a thread to ensure that the designated
65+ -- queue is not empty.
66+ void $ forkIO $ waiter tvar sq outputQ
67+ atomically $ writeTVar tvar (SyncNext out)
6468 let push b = do
6569 atomically $ writeTBQueue sq (SBuilder b)
6670 T.tickle th
6771 flush = atomically $ writeTBQueue sq SFlush
68− -- Since we must not enqueue an empty queue to the priority
69− -- queue, we spawn a thread to ensure that the designated
70− -- queue is not empty.
71− void $ forkIO $ waiter tvar sq (enqueue outputQ) strm
7272 strmbdy push flush
7373 atomically $ writeTBQueue sq SFinish
7474 _ -> do
7575 setThreadContinue tconf True
7676 let hasBody = requestMethod req /= H.methodHead
7777 && R.hasBody (responseStatus rsp)
78− pri <- readIORef $ streamPriority strm
79− enqueue outputQ (OResponse strm rsp (Oneshot hasBody)) pri
78+ out = OResponse strm rsp (Oneshot hasBody)
79+ sw <- atomically $ readTVar $ streamWindow strm
80+ if sw == 0 then
81+ void $ forkIO $ enqueueWhenWindowIsOpen outputQ out
82+ else do
83+ pri <- readIORef $ streamPriority strm
84+ enqueue outputQ out pri
8085 return ResponseReceived
8186
8287data Break = Break deriving (Show, Typeable)
@@ -128,27 +133,27 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
128133 Just e -> S.settingsOnException set (Just req) e
129134 clearStreamInfo sinfo
130135
131−waiter :: TVar Sync -> TBQueue Sequence
132− -> (Output -> Priority -> IO ()) -> Stream
133− -> IO ()
134−waiter tvar sq enq strm = do
136+waiter :: TVar Sync -> TBQueue Sequence -> PriorityTree Output -> IO ()
137+waiter tvar sq outQ = do
138+ -- waiting for actions other than SyncNone
135139 mx <- atomically $ do
136140 mout <- readTVar tvar
137141 case mout of
138142 SyncNone -> retry
139− SyncNext nxt -> do
143+ SyncNext out -> do
140144 writeTVar tvar SyncNone
141− return $ Just nxt
145+ return $ Just out
142146 SyncFinish -> return Nothing
143147 case mx of
144148 Nothing -> return ()
145− Just next -> do
149+ Just out -> do
150+ -- ensuring that the streaming queue is not empty.
146151 atomically $ do
147152 isEmpty <- isEmptyTBQueue sq
148153 when isEmpty retry
149− pri <- readIORef $ streamPriority strm
150− enq (ONext strm next) pri
151− waiter tvar sq enq strm
154+ -- ensuring that stream window is greater than 0.
155+ enqueueWhenWindowIsOpen outQ out
156+ waiter tvar sq outQ
152157
153158----------------------------------------------------------------
154159