http2 5.2.2 → 5.2.3
raw patch · 6 files changed
+72/−49 lines, 6 filesdep ~containersdep ~http-semanticsdep ~network-run
Dependency ranges changed: containers, http-semantics, network-run
Files
- ChangeLog.md +7/−0
- Network/HTTP2/Client/Run.hs +16/−18
- Network/HTTP2/H2/Receiver.hs +3/−1
- Network/HTTP2/H2/Sender.hs +41/−25
- Network/HTTP2/Server/Worker.hs +1/−1
- http2.cabal +4/−4
ChangeLog.md view
@@ -1,3 +1,10 @@+## 5.2.3++* Update for latest http-semantics+ [#120](https://github.com/kazu-yamamoto/http2/pull/120)+* Enable containers 0.7 (ghc 9.10)+ [#117](https://github.com/kazu-yamamoto/http2/pull/117)+ ## 5.2.2 * Mark final chunk as final
Network/HTTP2/Client/Run.hs view
@@ -6,7 +6,6 @@ module Network.HTTP2.Client.Run where import Control.Concurrent.STM (check)-import Data.ByteString.Builder (Builder) import qualified Data.ByteString.UTF8 as UTF8 import Data.IORef import Network.Control (RxFlow (..), defaultMaxData)@@ -170,11 +169,6 @@ -- the ordering of responses can be out-of-order. -- But for clients, the ordering must be maintained. -- To implement this, 'outputQStreamID' is used.- -- Also, for 'OutBodyStreaming', TBQ must not be empty- -- when its 'Output' is enqueued into 'outputQ'.- -- Otherwise, it would be re-enqueue because of empty- -- resulting in out-of-order.- -- To implement this, 'tbqNonEmpty' is used. let hdr1, hdr2 :: [Header] hdr1 | scheme /= "" = (":scheme", scheme) : hdr0@@ -187,8 +181,8 @@ (sid, newstrm) <- openOddStreamWait ctx case outObjBody req of OutBodyStreaming strmbdy ->- sendStreaming ctx mgr req' sid newstrm $ \unmask push flush ->- unmask $ strmbdy push flush+ sendStreaming ctx mgr req' sid newstrm $ \iface ->+ outBodyUnmask iface $ strmbdy (outBodyPush iface) (outBodyFlush iface) OutBodyStreamingUnmask strmbdy -> sendStreaming ctx mgr req' sid newstrm strmbdy _ -> atomically $ do@@ -204,23 +198,27 @@ -> OutObj -> StreamId -> Stream- -> ((forall x. IO x -> IO x) -> (Builder -> IO ()) -> IO () -> IO ())+ -> (OutBodyIface -> IO ()) -> IO () sendStreaming Context{..} mgr req sid newstrm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10- tbqNonEmpty <- newTVarIO False forkManagedUnmask mgr $ \unmask -> do- let push b = atomically $ do- writeTBQueue tbq (StreamingBuilder b)- writeTVar tbqNonEmpty True- flush = atomically $ writeTBQueue tbq StreamingFlush- finished = atomically $ writeTBQueue tbq $ StreamingFinished (decCounter mgr)+ decrementedCounter <- newIORef False+ let decCounterOnce = do+ alreadyDecremented <- atomicModifyIORef decrementedCounter $ \b -> (True, b)+ unless alreadyDecremented $ decCounter mgr+ let iface = OutBodyIface {+ outBodyUnmask = unmask+ , outBodyPush = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b Nothing)+ , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b (Just decCounterOnce))+ , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush+ }+ finished = atomically $ writeTBQueue tbq $ StreamingFinished decCounterOnce incCounter mgr- strmbdy unmask push flush `finally` finished+ strmbdy iface `finally` finished atomically $ do sidOK <- readTVar outputQStreamID- ready <- readTVar tbqNonEmpty- check (sidOK == sid && ready)+ check (sidOK == sid) writeTVar outputQStreamID (sid + 2) writeTQueue outputQ $ Output newstrm req OObj (Just tbq) (return ())
Network/HTTP2/H2/Receiver.hs view
@@ -620,7 +620,9 @@ (IORef Bool) mkSource- :: TQueue (Either E.SomeException (ByteString, Bool)) -> (Int -> IO ()) -> IO Source+ :: TQueue (Either E.SomeException (ByteString, Bool))+ -> (Int -> IO ())+ -> IO Source mkSource q inform = Source inform q <$> newIORef "" <*> newIORef False readSource :: Source -> IO (ByteString, Bool)
Network/HTTP2/H2/Sender.hs view
@@ -157,7 +157,7 @@ let payloadOff = off0 + frameHeaderLength datBuf = confWriteBuffer `plusPtr` payloadOff datBufSiz = buflim - payloadOff- Next datPayloadLen reqflush mnext <- curr datBuf datBufSiz lim -- checkme+ Next datPayloadLen reqflush mnext <- curr datBuf (min datBufSiz lim) NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen fillDataHeaderEnqueueNext strm@@ -168,7 +168,26 @@ sentinel out reqflush- output out@(Output strm (OutObj hdr body tlrmkr) OObj mtbq _) off0 lim = do+ output (Output strm obj OObj mtbq sentinel) off0 _lim = do+ outputObj strm obj mtbq sentinel off0+ output out@(Output strm _ (OPush ths pid) _ _) off0 lim = do+ -- Creating a push promise header+ -- Frame id should be associated stream id from the client.+ let sid = streamNumber strm+ len <- pushPromise pid sid ths off0+ off <- flushIfNecessary $ off0 + frameHeaderLength + len+ output out{outputType = OObj} off lim+ output _ _ _ = undefined -- never reach++ ----------------------------------------------------------------+ outputObj+ :: Stream+ -> OutObj+ -> Maybe (TBQueue StreamingChunk)+ -> IO ()+ -> Offset+ -> IO Offset+ outputObj strm obj@(OutObj hdr body tlrmkr) mtbq sentinel off0 = do -- Header frame and Continuation frame let sid = streamNumber strm endOfStream = case body of@@ -180,6 +199,7 @@ -- the stream from stream table. when endOfStream $ halfClosedLocal ctx strm Finished off <- flushIfNecessary off'+ let setOutputType otyp = Output strm obj otyp mtbq sentinel case body of OutBodyNone -> return off OutBodyFile (FileSpec path fileoff bytecount) -> do@@ -188,40 +208,33 @@ Closer closer -> timeoutClose mgr closer Refresher refresher -> return refresher let next = fillFileBodyGetNext pread fileoff bytecount refresh- out' = out{outputType = ONext next tlrmkr}- output out' off lim+ out' = setOutputType $ ONext next tlrmkr+ outputOrEnqueueAgain out' off OutBodyBuilder builder -> do let next = fillBuilderBodyGetNext builder- out' = out{outputType = ONext next tlrmkr}- output out' off lim- OutBodyStreaming _ ->- output (setNextForStreaming mtbq tlrmkr out) off lim- OutBodyStreamingUnmask _ ->- output (setNextForStreaming mtbq tlrmkr out) off lim- output out@(Output strm _ (OPush ths pid) _ _) off0 lim = do- -- Creating a push promise header- -- Frame id should be associated stream id from the client.- let sid = streamNumber strm- len <- pushPromise pid sid ths off0- off <- flushIfNecessary $ off0 + frameHeaderLength + len- output out{outputType = OObj} off lim- output _ _ _ = undefined -- never reach+ out' = setOutputType $ ONext next tlrmkr+ outputOrEnqueueAgain out' off+ OutBodyStreaming _ -> do+ let out' = setOutputType $ nextForStreaming mtbq tlrmkr+ outputOrEnqueueAgain out' off+ OutBodyStreamingUnmask _ -> do+ let out' = setOutputType $ nextForStreaming mtbq tlrmkr+ outputOrEnqueueAgain out' off ----------------------------------------------------------------- setNextForStreaming+ nextForStreaming :: Maybe (TBQueue StreamingChunk) -> TrailersMaker- -> Output Stream- -> Output Stream- setNextForStreaming mtbq tlrmkr out =+ -> OutputType+ nextForStreaming mtbq tlrmkr = let tbq = fromJust mtbq takeQ = atomically $ tryReadTBQueue tbq next = fillStreamBodyGetNext takeQ- in out{outputType = ONext next tlrmkr}+ in ONext next tlrmkr ---------------------------------------------------------------- outputOrEnqueueAgain :: Output Stream -> Offset -> IO Offset- outputOrEnqueueAgain out@(Output strm _ otyp _ _) off = E.handle resetStream $ do+ outputOrEnqueueAgain out@(Output strm obj otyp mtbq sentinel) off = E.handle resetStream $ do state <- readStreamState strm if isHalfClosedLocal state then return off@@ -230,11 +243,14 @@ -- Checking if all push are done. forkAndEnqueueWhenReady wait outputQ out{outputType = OObj} mgr return off+ OObj ->+ -- Send headers immediately, without waiting for data+ -- No need to check the streaming window (applies to DATA frames only)+ outputObj strm obj mtbq sentinel off _ -> case mtbq of Just tbq -> checkStreaming tbq _ -> checkStreamWindowSize where- mtbq = outputStrmQ out checkStreaming tbq = do isEmpty <- atomically $ isEmptyTBQueue tbq if isEmpty
Network/HTTP2/Server/Worker.hs view
@@ -150,7 +150,7 @@ writeOutputQ $ Output strm rsp otyp (Just tbq) (return ()) let push b = do T.pause th- atomically $ writeTBQueue tbq (StreamingBuilder b)+ atomically $ writeTBQueue tbq (StreamingBuilder b Nothing) T.resume th flush = atomically $ writeTBQueue tbq StreamingFlush finished = atomically $ writeTBQueue tbq $ StreamingFinished (decCounter mgr)
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 5.2.2+version: 5.2.3 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>@@ -114,8 +114,8 @@ async >=2.2 && <2.3, bytestring >=0.10, case-insensitive >=1.2 && <1.3,- containers >=0.6 && <0.7,- http-semantics >= 0.0.1,+ containers >=0.6,+ http-semantics >= 0.1 && <0.2, http-types >=0.12 && <0.13, network >=3.1, network-byte-order >=0.1.7 && <0.2,@@ -139,7 +139,7 @@ bytestring, http-types, http2,- network-run,+ network-run >= 0.3 && <0.4, unliftio if flag(devel)