diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/Network/HTTP2/Client/Run.hs b/Network/HTTP2/Client/Run.hs --- a/Network/HTTP2/Client/Run.hs +++ b/Network/HTTP2/Client/Run.hs @@ -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 ()) diff --git a/Network/HTTP2/H2/Receiver.hs b/Network/HTTP2/H2/Receiver.hs --- a/Network/HTTP2/H2/Receiver.hs +++ b/Network/HTTP2/H2/Receiver.hs @@ -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) diff --git a/Network/HTTP2/H2/Sender.hs b/Network/HTTP2/H2/Sender.hs --- a/Network/HTTP2/H2/Sender.hs +++ b/Network/HTTP2/H2/Sender.hs @@ -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 diff --git a/Network/HTTP2/Server/Worker.hs b/Network/HTTP2/Server/Worker.hs --- a/Network/HTTP2/Server/Worker.hs +++ b/Network/HTTP2/Server/Worker.hs @@ -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) diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -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 @@ -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)