http2 4.1.4 → 4.2.0

raw patch · 23 files changed

+349/−161 lines, 23 filesdep +randomPVP ok

version bump matches the API change (PVP)

Dependencies added: random

API changes (from Hackage documentation)

Files

• ChangeLog.md +21/−5
• Network/HTTP2/Arch/Config.hs +8/−0
• Network/HTTP2/Arch/Context.hs +13/−7
• Network/HTTP2/Arch/Manager.hs +60/−22
• Network/HTTP2/Arch/Receiver.hs +31/−35
• Network/HTTP2/Arch/Sender.hs +28/−14
• Network/HTTP2/Arch/Stream.hs +22/−3
• Network/HTTP2/Arch/Types.hs +23/−7
• Network/HTTP2/Client.hs +9/−0
• Network/HTTP2/Client/Run.hs +44/−24
• Network/HTTP2/Client/Types.hs +3/−1
• Network/HTTP2/Internal.hs +5/−2
• Network/HTTP2/Server.hs +3/−1
• Network/HTTP2/Server/Run.hs +9/−3
• Network/HTTP2/Server/Types.hs +7/−2
• Network/HTTP2/Server/Worker.hs +12/−9
• http2.cabal +3/−2
• test-hpack/hpack-encode.hs +6/−5
• test-hpack/hpack-stat.hs +4/−3
• test/HTTP2/ClientSpec.hs +5/−1
• test/HTTP2/ServerSpec.hs +21/−6
• util/client.hs +7/−5
• util/server.hs +5/−4

@@ -1,9 +1,25 @@-## 4.1.4+## 4.2.0 -* Handle RST_STREAM/NO_ERROR-  [#78](https://github.com/kazu-yamamoto/http2/issues/78)-* Fixing thread leak with forkManaged-  [#74](https://github.com/kazu-yamamoto/http2/issues/74)+* Treating HALF_CLOSED_LOCAL correctly.+  [#90](https://github.com/kazu-yamamoto/http2/pull/90)+* Ensuring that GOAWAY is sent after DATA in the client side.+  [#89](https://github.com/kazu-yamamoto/http2/pull/90)+* Test uses a random port instead of 8080.+* Breaking change: adding two optional `SockAddr`s to `Config` to be copied into `Aux`.+* Close all streams on termination.+  [#83](https://github.com/kazu-yamamoto/http2/pull/83)+* Introducing `OutBodyStreamingUnmask`+  [#80](https://github.com/kazu-yamamoto/http2/pull/80)+* Introducing `KilledByHttp2ThreadManager` instead of `ThreadKilled`.+  [#79](https://github.com/kazu-yamamoto/http2/pull/79)+  [#81](https://github.com/kazu-yamamoto/http2/pull/82)+  [#82](https://github.com/kazu-yamamoto/http2/pull/82)+* Handle RST_STREAM with NO_ERROR.+  [#78](https://github.com/kazu-yamamoto/http2/pull/78)+* Internal changes:+  [#74](https://github.com/kazu-yamamoto/http2/pull/74)+* Breaking change: `Client` is generalized into `(forall b. Request -> (Response -> IO b) -> IO b) -> IO a`. The `RankNTypes` language extension is required.+  [#72](https://github.com/kazu-yamamoto/http2/pull/72)  ## 4.1.3 

@@ -25,6 +25,10 @@     , confReadN       :: Int -> IO ByteString     , confPositionReadMaker :: PositionReadMaker     , confTimeoutManager :: T.Manager+    -- | This is copied into 'Aux', if exist, on server.+    , confMySockAddr     :: SockAddr+    -- | This is copied into 'Aux', if exist, on server.+    , confPeerSockAddr   :: SockAddr     }  -- | Making simple configuration whose IO is not efficient.@@ -34,6 +38,8 @@     buf <- mallocBytes bufsiz     ref <- newIORef Nothing     timmgr <- T.initialize $ 30 * 1000000+    mysa <- getSocketName s+    peersa <- getPeerName s     let config = Config {             confWriteBuffer = buf           , confBufferSize = bufsiz@@ -41,6 +47,8 @@           , confReadN = defaultReadN s ref           , confPositionReadMaker = defaultPositionReadMaker           , confTimeoutManager = timmgr+          , confMySockAddr   = mysa+          , confPeerSockAddr = peersa           }     return config 

@@ -4,6 +4,7 @@  import Data.IORef import Network.HTTP.Types (Method)+import Network.Socket (SockAddr) import UnliftIO.STM  import Imports hiding (insert)@@ -87,12 +88,14 @@   , pingRate           :: Rate   , settingsRate       :: Rate   , emptyFrameRate     :: Rate+  , mySockAddr         :: SockAddr+  , peerSockAddr       :: SockAddr   }  ---------------------------------------------------------------- -newContext :: RoleInfo -> BufferSize -> IO Context-newContext rinfo siz =+newContext :: RoleInfo -> BufferSize -> SockAddr -> SockAddr -> IO Context+newContext rinfo siz mysa peersa =     Context rl rinfo                <$> newIORef False                <*> newIORef Nothing@@ -115,6 +118,8 @@                <*> newRate                <*> newRate                <*> newRate+               <*> return mysa+               <*> return peersa    where      rl = case rinfo of        RIC{} -> Client@@ -155,7 +160,7 @@ opened :: Context -> Stream -> IO () opened ctx@Context{concurrency} strm = do     atomicModifyIORef' concurrency (\x -> (x+1,()))-    setStreamState ctx strm (Open JustOpened)+    setStreamState ctx strm (Open Nothing JustOpened)  halfClosedRemote :: Context -> Stream -> IO () halfClosedRemote ctx stream@Stream{streamState} = do@@ -163,9 +168,9 @@     traverse_ (closed ctx stream) closingCode   where     closeHalf :: StreamState -> (StreamState, Maybe ClosedCode)-    closeHalf x@(Closed _)         = (x, Nothing)-    closeHalf (HalfClosedLocal cc) = (Closed cc, Just cc)-    closeHalf _                    = (HalfClosedRemote, Nothing)+    closeHalf x@(Closed _)       = (x, Nothing)+    closeHalf (Open (Just cc) _) = (Closed cc, Just cc)+    closeHalf _                  = (HalfClosedRemote, Nothing)  halfClosedLocal :: Context -> Stream -> ClosedCode -> IO () halfClosedLocal ctx stream@Stream{streamState} cc = do@@ -176,7 +181,8 @@     closeHalf :: StreamState -> (StreamState, Bool)     closeHalf x@(Closed _)     = (x, False)     closeHalf HalfClosedRemote = (Closed cc, True)-    closeHalf _                = (HalfClosedLocal cc, False)+    closeHalf (Open Nothing o) = (Open (Just cc) o, False)+    closeHalf _                = (Open (Just cc) JustOpened, False)  closed :: Context -> Stream -> ClosedCode -> IO () closed ctx@Context{concurrency,streamTable} strm@Stream{streamNumber} cc = do

@@ -1,4 +1,6 @@--- | A thread pool manager.+{-# LANGUAGE RankNTypes #-}++-- | A thread manager. --   The manager has responsibility to spawn and kill --   worker threads. module Network.HTTP2.Arch.Manager (@@ -6,12 +8,16 @@   , Action   , start   , setAction-  , stop+  , stopAfter   , spawnAction   , forkManaged-  , deleteMyId+  , forkManagedUnmask   , timeoutKillThread   , timeoutClose+  , KilledByHttp2ThreadManager(..)+  , incCounter+  , decCounter+  , waitCounter0   ) where  import Control.Exception@@ -34,12 +40,12 @@ noAction :: Action noAction = return () -data Command = Stop | Spawn | Add ThreadId | Delete ThreadId+data Command = Stop (Maybe SomeException) | Spawn | Add ThreadId | Delete ThreadId --- | Manager to manage the thread pool and the timer.-data Manager = Manager (TQueue Command) (IORef Action) T.Manager+-- | Manager to manage the thread and the timer.+data Manager = Manager (TQueue Command) (IORef Action) (TVar Int) T.Manager --- | Starting a thread pool manager.+-- | Starting a thread manager. --   Its action is initially set to 'return ()' and should be set --   by 'setAction'. This allows that the action can include --   the manager itself.@@ -47,13 +53,14 @@ start timmgr = do     q <- newTQueueIO     ref <- newIORef noAction+    cnt <- newTVarIO 0     void $ forkIO $ go q Set.empty ref-    return $ Manager q ref timmgr+    return $ Manager q ref cnt timmgr   where     go q tset0 ref = do         x <- atomically $ readTQueue q         case x of-            Stop          -> kill tset0+            Stop err      -> kill tset0 err             Spawn         -> next tset0             Add    newtid -> let tset = add newtid tset0                              in go q tset ref@@ -62,21 +69,27 @@       where         next tset = do             action <- readIORef ref-            newtid <- forkIO action+            newtid <- forkFinally action $ \_ -> do+                mytid <- myThreadId+                atomically $ writeTQueue q $ Delete mytid             let tset' = add newtid tset             go q tset' ref  -- | Setting the action to be spawned. setAction :: Manager -> Action -> IO ()-setAction (Manager _ ref _) action = writeIORef ref action+setAction (Manager _ ref _ _) action = writeIORef ref action  -- | Stopping the manager.-stop :: Manager -> IO ()-stop (Manager q _ _) = atomically $ writeTQueue q Stop+stopAfter :: Manager -> IO a -> (Either SomeException a -> IO b) -> IO b+stopAfter (Manager q _ _ _) action cleanup = do+   mask $ \unmask -> do+     ma <- try $ unmask action+     atomically $ writeTQueue q $ Stop (either Just (const Nothing) ma)+     cleanup ma  -- | Spawning the action. spawnAction :: Manager -> IO ()-spawnAction (Manager q _ _) = atomically $ writeTQueue q Spawn+spawnAction (Manager q _ _ _) = atomically $ writeTQueue q Spawn  ---------------------------------------------------------------- @@ -87,9 +100,14 @@ -- (normally or abnormally). forkManaged :: Manager -> IO () -> IO () forkManaged mgr io =+    forkManagedUnmask mgr $ \unmask -> unmask io++-- | Like 'forkManaged', but run action with exceptions masked+forkManagedUnmask :: Manager -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()+forkManagedUnmask mgr io =     void $ mask_ $ forkIOWithUnmask $ \unmask -> do       addMyId mgr-      r <- unmask io `onException` deleteMyId mgr+      r <- io unmask `onException` deleteMyId mgr       deleteMyId mgr       return r @@ -97,7 +115,7 @@ -- -- This is not part of the public API; see 'forkManaged' instead. addMyId :: Manager -> IO ()-addMyId (Manager q _ _) = do+addMyId (Manager q _ _ _) = do     tid <- myThreadId     atomically $ writeTQueue q $ Add tid @@ -107,7 +125,7 @@ -- the manager /before/ the thread terminates (thereby assuming responsibility -- for thread cleanup yourself). deleteMyId :: Manager -> IO ()-deleteMyId (Manager q _ _) = do+deleteMyId (Manager q _ _ _) = do     tid <- myThreadId     atomically $ writeTQueue q $ Delete tid @@ -123,18 +141,38 @@   where     set' = Set.delete tid set -kill :: Set ThreadId -> IO ()-kill set = traverse_ killThread set+kill :: Set ThreadId -> Maybe SomeException -> IO ()+kill set err = traverse_ (\tid -> E.throwTo tid $ KilledByHttp2ThreadManager err) set  -- | Killing the IO action of the second argument on timeout.-timeoutKillThread :: Manager -> (T.Handle -> IO ()) -> IO ()-timeoutKillThread (Manager _ _ tmgr) action = E.bracket register T.cancel action+timeoutKillThread :: Manager -> (T.Handle -> IO a) -> IO a+timeoutKillThread (Manager _ _ _ tmgr) action = E.bracket register T.cancel action   where     register = T.registerKillThread tmgr noAction  -- | Registering closer for a resource and --   returning a timer refresher. timeoutClose :: Manager -> IO () -> IO (IO ())-timeoutClose (Manager _ _ tmgr) closer = do+timeoutClose (Manager _ _ _ tmgr) closer = do     th <- T.register tmgr closer     return $ T.tickle th++data KilledByHttp2ThreadManager = KilledByHttp2ThreadManager (Maybe SomeException)+  deriving Show++instance Exception KilledByHttp2ThreadManager where+  toException   = asyncExceptionToException+  fromException = asyncExceptionFromException++----------------------------------------------------------------++incCounter :: Manager -> IO ()+incCounter (Manager _ _ cnt _) = atomically $ modifyTVar' cnt (+1)++decCounter :: Manager -> IO ()+decCounter (Manager _ _ cnt _) = atomically $ modifyTVar' cnt (subtract 1)++waitCounter0 :: Manager -> IO ()+waitCounter0 (Manager _ _ cnt _) = atomically $ do+    n <- readTVar cnt+    checkSTM (n < 1)

@@ -174,7 +174,7 @@ processState :: StreamState -> Context -> Stream -> StreamId -> IO Bool  -- Transition (process1)-processState (Open (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} streamId = do+processState (Open _ (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} streamId = do     let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt)     when (just mcl (/= (0 :: Int))) $ E.throwIO $ StreamErrorIsSent ProtocolError streamId "no body but content-length is not zero"     halfClosedRemote ctx strm@@ -188,12 +188,12 @@     return False  -- Transition (process2)-processState (Open (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do+processState (Open hcl (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do     let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt)     bodyLength <- newIORef 0     tlr <- newIORef Nothing     q <- newTQueueIO-    setStreamState ctx strm $ Open (Body q mcl bodyLength tlr)+    setStreamState ctx strm $ Open hcl (Body q mcl bodyLength tlr)     incref <- newIORef 0     bodySource <- mkSource q $ informWindowUpdate ctx strm incref     let inpObj = InpObj tbl mcl (readSource bodySource) tlr@@ -205,7 +205,7 @@     return False  -- Transition (process3)-processState s@(Open Continued{}) ctx strm _streamId = do+processState s@(Open _ Continued{}) ctx strm _streamId = do     setStreamState ctx strm s     return True @@ -343,7 +343,7 @@ stream :: FrameType -> FrameHeader -> ByteString -> Context -> StreamState -> Stream -> IO StreamState  -- Transition (stream1)-stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx s@(Open JustOpened) Stream{streamNumber} = do+stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx s@(Open hcl JustOpened) Stream{streamNumber} = do     HeadersFrame mp frag <- guardIt $ decodeHeadersFrame header bs     let endOfStream = testEndStream flags         endOfHeader = testEndHeader flags@@ -361,44 +361,33 @@         if endOfHeader then do             tbl <- hpackDecodeHeader frag streamId ctx             return $ if endOfStream then-                        Open (NoBody tbl)+                       -- turned into HalfClosedRemote in processState+                        Open hcl (NoBody tbl)                        else-                        Open (HasBody tbl)+                        Open hcl (HasBody tbl)           else do             let siz = BS.length frag-            return $ Open $ Continued [frag] siz 1 endOfStream+            return $ Open hcl $ Continued [frag] siz 1 endOfStream  -- Transition (stream2)-stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open (Body q _ _ tlr)) _ = do+stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open _ (Body q _ _ tlr)) _ = do     HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs     let endOfStream = testEndStream flags     -- checking frag == "" is not necessary     if endOfStream then do         tbl <- hpackDecodeTrailer frag streamId ctx         writeIORef tlr (Just tbl)-        atomically $ writeTQueue q ""+        atomically $ writeTQueue q $ Right ""         return HalfClosedRemote       else         -- we don't support continuation here.         E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "continuation in trailer is not supported" --- ignore data-frame except for flow-control when we're done locally-stream FrameData-       FrameHeader{flags}-       _bs-       _ctx s@(HalfClosedLocal _)-       _ = do-    let endOfStream = testEndStream flags-    if endOfStream then do-        return HalfClosedRemote-      else-        return s- -- Transition (stream4) stream FrameData        header@FrameHeader{flags,payloadLength,streamId}        bs-       Context{emptyFrameRate} s@(Open (Body q mcl bodyLength _))+       Context{emptyFrameRate} s@(Open _ (Body q mcl bodyLength _))        _ = do     DataFrame body <- guardIt $ decodeDataFrame header bs     len0 <- readIORef bodyLength@@ -412,19 +401,19 @@                 E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "too many empty data"       else do         writeIORef bodyLength len-        atomically $ writeTQueue q body+        atomically $ writeTQueue q $ Right body     if endOfStream then do         case mcl of             Nothing -> return ()             Just cl -> when (cl /= len) $ E.throwIO $ StreamErrorIsSent ProtocolError streamId "actual body length is not the same as content-length"         -- no trailers-        atomically $ writeTQueue q ""+        atomically $ writeTQueue q $ Right ""         return HalfClosedRemote       else         return s  -- Transition (stream5)-stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open (Continued rfrags siz n endOfStream)) _ = do+stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open hcl (Continued rfrags siz n endOfStream)) _ = do     let endOfHeader = testEndHeader flags     if frag == "" && not endOfHeader then do         -- Empty Frame Flooding - CVE-2019-9518@@ -445,11 +434,12 @@             let hdrblk = BS.concat $ reverse rfrags'             tbl <- hpackDecodeHeader hdrblk streamId ctx             return $ if endOfStream then-                        Open (NoBody tbl)+                        -- turned into HalfClosedRemote in processState+                        Open hcl (NoBody tbl)                        else-                        Open (HasBody tbl)+                        Open hcl (HasBody tbl)           else-            return $ Open $ Continued rfrags' siz' n' endOfStream+            return $ Open hcl $ Continued rfrags' siz' n' endOfStream  -- (No state transition) stream FrameWindowUpdate header bs _ s strm = do@@ -488,7 +478,7 @@  -- this ordering is important stream FrameContinuation FrameHeader{streamId} _ _ _ _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "continue frame cannot come here"-stream _ FrameHeader{streamId} _ _ (Open Continued{}) _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "an illegal frame follows header/continuation frames"+stream _ FrameHeader{streamId} _ _ (Open _ Continued{}) _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "an illegal frame follows header/continuation frames" -- Ignore frames to streams we have just reset, per section 5.1. stream _ _ _ _ st@(Closed (ResetByMe _)) _ = return st stream FrameData FrameHeader{streamId} _ _ _ _ = E.throwIO $ StreamErrorIsSent StreamClosed streamId $ fromString ("illegal data frame for " ++ show streamId)@@ -498,11 +488,11 @@  -- | Type for input streaming. data Source = Source (Int -> IO ())-                     (TQueue ByteString)+                     (TQueue (Either E.SomeException ByteString))                      (IORef ByteString)                      (IORef Bool) -mkSource :: TQueue ByteString -> (Int -> IO ()) -> IO Source+mkSource :: TQueue (Either E.SomeException ByteString) -> (Int -> IO ()) -> IO Source mkSource q inform = Source inform q <$> newIORef "" <*> newIORef False  readSource :: Source -> IO ByteString@@ -516,12 +506,18 @@         inform len         return bs   where+    readBS :: IO ByteString     readBS = do         bs0 <- readIORef refBS         if bs0 == "" then do-            bs <- atomically $ readTQueue q-            when (bs == "") $ writeIORef refEOF True-            return bs+            mBS <- atomically $ readTQueue q+            case mBS of+              Left err -> do+                writeIORef refEOF True+                E.throwIO err+              Right bs -> do+                when (bs == "") $ writeIORef refEOF True+                return bs           else do             writeIORef refBS ""             return bs0

@@ -11,6 +11,7 @@   , runTrailersMaker   ) where +import Control.Concurrent.MVar (putMVar) import qualified Data.ByteString as BS import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Builder.Extra as B@@ -106,6 +107,12 @@     -- called with off == 0     control :: Control -> IO ()     control (CFinish     e) = E.throwIO e+    control (CGoaway bs mvar) = do+        buf <- copyAll [bs] confWriteBuffer+        let off = buf `minusPtr` confWriteBuffer+        flushN off+        putMVar mvar ()+        E.throwIO GoAwayIsSent     control (CFrames ms xs) = do         buf <- copyAll xs confWriteBuffer         let off = buf `minusPtr` confWriteBuffer@@ -148,13 +155,12 @@                 _           -> False         (ths,_) <- toHeaderTable $ fixHeaders hdr         off' <- headerContinue sid ths endOfStream off0+        -- halfClosedLocal calls closed which removes+        -- the stream from stream table.+        when endOfStream $ halfClosedLocal ctx strm Finished         off <- flushIfNecessary off'         case body of-            OutBodyNone -> do-                -- halfClosedLocal calls closed which removes-                -- the stream from stream table.-                when (isServer ctx) $ halfClosedLocal ctx strm Finished-                return off+            OutBodyNone -> return off             OutBodyFile (FileSpec path fileoff bytecount) -> do                 (pread, sentinel') <- confPositionReadMaker path                 refresh <- case sentinel' of@@ -167,12 +173,10 @@                 let next = fillBuilderBodyGetNext builder                     out' = out { outputType = ONext next tlrmkr }                 output out' off lim-            OutBodyStreaming _ -> do-                let tbq = fromJust mtbq-                    takeQ = atomically $ tryReadTBQueue tbq-                    next = fillStreamBodyGetNext takeQ-                    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@@ -185,6 +189,14 @@     output _ _ _ = undefined -- never reach      ----------------------------------------------------------------+    setNextForStreaming :: Maybe (TBQueue StreamingChunk) -> TrailersMaker -> Output Stream -> Output Stream+    setNextForStreaming mtbq tlrmkr out =+        let tbq = fromJust mtbq+            takeQ = atomically $ tryReadTBQueue tbq+            next = fillStreamBodyGetNext takeQ+        in out { outputType = ONext next tlrmkr }++    ----------------------------------------------------------------     outputOrEnqueueAgain :: Output Stream -> Offset -> IO Offset     outputOrEnqueueAgain out@(Output strm _ otyp _ _) off = E.handle resetStream $ do         state <- readStreamState strm@@ -282,7 +294,7 @@         fillFrameHeader FrameData datPayloadLen streamNumber flag buf         off'' <- handleTrailers mtrailers off'         void tell-        when (isServer ctx) $ halfClosedLocal ctx strm Finished+        halfClosedLocal ctx strm Finished         decreaseWindowSize ctx strm datPayloadLen         if reqflush then do             flushN off''@@ -293,7 +305,7 @@         handleTrailers Nothing off0 = return off0         handleTrailers (Just trailers) off0 = do             (ths,_) <- toHeaderTable trailers-            headerContinue streamNumber ths True off0+            headerContinue streamNumber ths True {- endOfStream -} off0      fillDataHeaderEnqueueNext _                    off 0 (Just next) tlrmkr _ out reqflush = do@@ -426,7 +438,9 @@                     B.More  _ writer  -> return (True,  total', False, LOne writer)                     B.Chunk bs writer -> return (True,  total', False, LTwo bs writer)             Just StreamingFlush       -> return (True,  total,  True,  LZero)-            Just StreamingFinished    -> return (False, total,  True,  LZero)+            Just (StreamingFinished dec) -> do+                dec+                return (False, total,  True,  LZero)  fillBufStream :: Leftover -> IO (Maybe StreamingChunk) -> DynaNext fillBufStream leftover0 takeQ buf0 siz0 lim0 = do

@@ -2,6 +2,7 @@  module Network.HTTP2.Arch.Stream where +import Control.Exception import Data.IORef import qualified Data.IntMap.Strict as M import UnliftIO.Concurrent@@ -27,9 +28,9 @@ isHalfClosedRemote _                = False  isHalfClosedLocal :: StreamState -> Bool-isHalfClosedLocal (HalfClosedLocal _) = True-isHalfClosedLocal (Closed _)       = True-isHalfClosedLocal _                = False+isHalfClosedLocal (Open (Just _) _) = True+isHalfClosedLocal (Closed _)        = True+isHalfClosedLocal _                 = False  isClosed :: StreamState -> Bool isClosed Closed{} = True@@ -75,3 +76,21 @@ updateAllStreamWindow adst (StreamTable ref) = do     strms <- M.elems <$> readIORef ref     forM_ strms $ \strm -> atomically $ modifyTVar (streamWindow strm) adst++closeAllStreams :: StreamTable -> Maybe SomeException -> IO ()+closeAllStreams (StreamTable ref) mErr' = do+    strms <- atomicModifyIORef' ref $ \m -> (M.empty, m)+    forM_ strms $ \strm -> do+      st <- readStreamState strm+      case st of+        Open _ (Body q _ _ _) ->+          atomically $ writeTQueue q $ maybe (Right mempty) Left mErr+        _otherwise ->+          return ()+  where+    mErr :: Maybe SomeException+    mErr = case mErr' of+             Just err | Just ConnectionIsClosed <- fromException err ->+               Nothing+             _otherwise ->+               mErr'

@@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}  module Network.HTTP2.Arch.Types where @@ -35,6 +36,20 @@ data OutBody = OutBodyNone              -- | Streaming body takes a write action and a flush action.              | OutBodyStreaming ((Builder -> IO ()) -> IO () -> IO ())+             -- | Like 'OutBodyStreaming', but with a callback to unmask expections+             --+             -- This is used in the client: we spawn the new thread for the request body+             -- with exceptions masked, and provide the body of 'OutBodyStreamingUnmask'+             -- with a callback to unmask them again (typically after installing an exception+             -- handler).+             --+             -- We do /NOT/ support this in the server, as here the scope of the thread+             -- that is spawned for the server is the entire handler, not just the response+             -- streaming body.+             --+             -- TODO: The analogous change for the server-side would be to provide a similar+             -- @unmask@ callback as the first argument in the 'Server' type alias.+             | OutBodyStreamingUnmask ((forall x. IO x -> IO x) -> (Builder -> IO ()) -> IO () -> IO ())              | OutBodyBuilder Builder              | OutBodyFile FileSpec @@ -189,7 +204,7 @@               Bool -- End of stream   | NoBody HeaderTable   | HasBody HeaderTable-  | Body (TQueue ByteString)+  | Body (TQueue (Either SomeException ByteString))          (Maybe Int) -- received Content-Length                      -- compared the body length for error checking          (IORef Int) -- actual body length@@ -205,17 +220,16 @@  data StreamState =     Idle-  | Open OpenState+  | Open (Maybe ClosedCode) OpenState -- HalfClosedLocal if Just   | HalfClosedRemote-  | HalfClosedLocal ClosedCode   | Closed ClosedCode   | Reserved  instance Show StreamState where     show Idle                = "Idle"-    show Open{}              = "Open"+    show (Open Nothing _)    = "Open"+    show (Open (Just e) _)   = "HalfClosedLocal: " ++ show e     show HalfClosedRemote    = "HalfClosedRemote"-    show (HalfClosedLocal e) = "HalfClosedLocal: " ++ show e     show (Closed e)          = "Closed: " ++ show e     show Reserved            = "Reserved" @@ -264,12 +278,13 @@  ---------------------------------------------------------------- -data Control = CFinish    HTTP2Error+data Control = CFinish HTTP2Error              | CFrames (Maybe SettingsList) [ByteString]+             | CGoaway ByteString (MVar ())  ---------------------------------------------------------------- -data StreamingChunk = StreamingFinished+data StreamingChunk = StreamingFinished (IO ())                     | StreamingFlush                     | StreamingBuilder Builder @@ -290,6 +305,7 @@   | StreamErrorIsReceived     ErrorCode StreamId   | StreamErrorIsSent         ErrorCode StreamId ReasonPhrase   | BadThingHappen E.SomeException+  | GoAwayIsSent   deriving (Show, Typeable)  instance E.Exception HTTP2Error

@@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}  -- | HTTP\/2 client library. --@@ -58,6 +59,7 @@   , requestNoBody   , requestFile   , requestStreaming+  , requestStreamingUnmask   , requestBuilder   -- ** Trailers maker   , TrailersMaker@@ -129,6 +131,13 @@   where     hdr' = addHeaders m p hdr +-- | Like 'requestStreaming', but run the action with exceptions masked+requestStreamingUnmask :: Method -> Path -> RequestHeaders+                       -> ((forall x. IO x -> IO x) -> (Builder -> IO ()) -> IO () -> IO ())+                       -> Request+requestStreamingUnmask m p hdr strmbdy = Request $ OutObj hdr' (OutBodyStreamingUnmask strmbdy) defaultTrailersMaker+  where+    hdr' = addHeaders m p hdr  addHeaders :: Method -> Path -> RequestHeaders -> RequestHeaders addHeaders m p hdr = (":method", m) : (":path", p) : hdr

@@ -1,18 +1,20 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-}  module Network.HTTP2.Client.Run where  import Control.Concurrent.STM (check)+import Control.Exception import UnliftIO.Async import UnliftIO.Concurrent-import qualified UnliftIO.Exception as E import UnliftIO.STM  import Imports import Network.HTTP2.Arch import Network.HTTP2.Client.Types import Network.HTTP2.Frame+import Data.ByteString.Builder (Builder)  -- | Client configuration data ClientConfig = ClientConfig {@@ -25,22 +27,30 @@ run :: ClientConfig -> Config -> Client a -> IO a run ClientConfig{..} conf@Config{..} client = do     clientInfo <- newClientInfo scheme authority cacheLimit-    ctx <- newContext clientInfo confBufferSize+    ctx <- newContext clientInfo confBufferSize confMySockAddr confPeerSockAddr     mgr <- start confTimeoutManager     let runBackgroundThreads = do             let runReceiver = frameReceiver ctx conf                 runSender   = frameSender   ctx conf mgr             concurrently_ runReceiver runSender     exchangeSettings conf ctx+    mvar <- newMVar ()     let runClient = do             x <- client $ sendRequest ctx mgr scheme authority+            waitCounter0 mgr             let frame = goawayFrame 0 NoError "graceful closing"-            enqueueControl (controlQ ctx) $ CFrames Nothing [frame]+            enqueueControl (controlQ ctx) $ CGoaway frame mvar+            takeMVar mvar             return x-    ex <- race runBackgroundThreads runClient `E.finally` stop mgr-    case ex of-      Left () -> undefined -- never reach-      Right x -> return x+    stopAfter mgr (race runBackgroundThreads runClient) $ \res -> do+      closeAllStreams (streamTable ctx) $ either Just (const Nothing) res+      case res of+        Left err ->+          throwIO err+        Right (Left ()) ->+          undefined -- never reach+        Right (Right x) ->+          return x  sendRequest :: Context -> Manager -> Scheme -> Authority -> Request -> (Response -> IO a) -> IO a sendRequest ctx@Context{..} mgr scheme auth (Request req) processResponse = do@@ -59,7 +69,7 @@           -- when its 'Output' is enqueued into 'outputQ'.           -- Otherwise, it would be re-enqueue because of empty           -- resulting in out-of-order.-          -- To implement this, 'tbqNonMmpty' is used.+          -- To implement this, 'tbqNonEmpty' is used.           let hdr1 | scheme /= "" = (":scheme", scheme) : hdr0                    | otherwise    = hdr0               hdr2 | auth /= "" = (":authority", auth) : hdr1@@ -68,22 +78,11 @@           sid <- getMyNewStreamId ctx           newstrm <- openStream ctx sid FrameHeaders           case outObjBody req of-            OutBodyStreaming strmbdy -> do-                tbq <- newTBQueueIO 10 -- fixme: hard coding: 10-                tbqNonMmpty <- newTVarIO False-                forkManaged mgr $ do-                    let push b = atomically $ do-                            writeTBQueue tbq (StreamingBuilder b)-                            writeTVar tbqNonMmpty True-                        flush  = atomically $ writeTBQueue tbq StreamingFlush-                    strmbdy push flush-                    atomically $ writeTBQueue tbq StreamingFinished-                atomically $ do-                    sidOK <- readTVar outputQStreamID-                    ready <- readTVar tbqNonMmpty-                    check (sidOK == sid && ready)-                    writeTVar outputQStreamID (sid + 2)-                    writeTQueue outputQ $ Output newstrm req' OObj (Just tbq) (return ())+            OutBodyStreaming strmbdy ->+                sendStreaming ctx mgr req' sid newstrm $ \unmask push flush ->+                    unmask $ strmbdy push flush+            OutBodyStreamingUnmask strmbdy ->+                sendStreaming ctx mgr req' sid newstrm strmbdy             _ -> atomically $ do                 sidOK <- readTVar outputQStreamID                 check (sidOK == sid)@@ -93,6 +92,27 @@       Just strm0 -> return strm0     rsp <- takeMVar $ streamInput strm     processResponse $ Response rsp++sendStreaming :: Context -> Manager -> OutObj -> StreamId -> Stream+              -> ((forall x. IO x -> IO x) -> (Builder -> IO ()) -> IO () -> 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)+        incCounter mgr+        strmbdy unmask push flush `finally` finished+    atomically $ do+        sidOK <- readTVar outputQStreamID+        ready <- readTVar tbqNonEmpty+        check (sidOK == sid && ready)+        writeTVar outputQStreamID (sid + 2)+        writeTQueue outputQ $ Output newstrm req OObj (Just tbq) (return ())  exchangeSettings :: Config -> Context -> IO () exchangeSettings conf ctx@Context{..} = do

@@ -1,3 +1,5 @@+{-# LANGUAGE RankNTypes #-}+ module Network.HTTP2.Client.Types where  import Network.HTTP2.Arch@@ -5,7 +7,7 @@ ----------------------------------------------------------------  -- | Client type.-type Client a = (Request -> (Response -> IO a) -> IO a) -> IO a+type Client a = (forall b. Request -> (Response -> IO b) -> IO b) -> IO a  -- | Request from client. newtype Request = Request OutObj deriving (Show)

@@ -24,8 +24,11 @@   , defaultTrailersMaker   , NextTrailersMaker(..)   , runTrailersMaker-  )  where+  -- * Thread Manager+  , module Network.HTTP2.Arch.Manager+  ) where  import Network.HTTP2.Arch.File-import Network.HTTP2.Arch.Types+import Network.HTTP2.Arch.Manager import Network.HTTP2.Arch.Sender+import Network.HTTP2.Arch.Types

@@ -15,7 +15,7 @@ -- > import Network.HTTP2.Server -- > -- > main :: IO ()--- > main = runTCPServer Nothing "80" $ \s _peer -> runHTTP2Server s+-- > main = runTCPServer Nothing "80" runHTTP2Server -- >   where -- >     runHTTP2Server s = E.bracket (allocSimpleConfig s 4096) -- >                                  freeSimpleConfig@@ -49,6 +49,8 @@   -- * Aux   , Aux   , auxTimeHandle+  , auxMySockAddr+  , auxPeerSockAddr   -- * Response   , Response   -- ** Creating response

@@ -4,13 +4,13 @@ module Network.HTTP2.Server.Run where  import UnliftIO.Async (concurrently_)-import qualified UnliftIO.Exception as E  import Imports import Network.HTTP2.Arch import Network.HTTP2.Frame import Network.HTTP2.Server.Types import Network.HTTP2.Server.Worker+import Control.Exception  ---------------------------------------------------------------- @@ -20,7 +20,7 @@     ok <- checkPreface     when ok $ do         serverInfo <- newServerInfo-        ctx <- newContext serverInfo confBufferSize+        ctx <- newContext serverInfo confBufferSize confMySockAddr confPeerSockAddr         -- Workers, worker manager and timer manager         mgr <- start confTimeoutManager         let wc = fromContext ctx@@ -33,7 +33,13 @@         replicateM_ 3 $ spawnAction mgr         let runReceiver = frameReceiver ctx conf             runSender   = frameSender   ctx conf mgr-        concurrently_ runReceiver runSender `E.finally` stop mgr+        stopAfter mgr (concurrently_ runReceiver runSender) $ \res -> do+          closeAllStreams (streamTable ctx) $ either Just (const Nothing) res+          case res of+            Left err ->+              throwIO err+            Right x ->+              return x   where     checkPreface = do         preface <- confReadN connectionPrefaceLength

@@ -1,5 +1,6 @@ module Network.HTTP2.Server.Types where +import Network.Socket (SockAddr) import qualified System.TimeManager as T  import Imports@@ -32,7 +33,11 @@     }  -- | Additional information.-newtype Aux = Aux {+data Aux = Aux {     -- | Time handle for the worker processing this request and response.-    auxTimeHandle :: T.Handle+    auxTimeHandle  :: T.Handle+    -- | Local socket address copied from 'Config'.+  , auxMySockAddr   :: SockAddr+    -- | Remove socket address copied from 'Config'.+  , auxPeerSockAddr :: SockAddr   }

@@ -9,9 +9,9 @@   , fromContext   ) where -import Control.Exception (AsyncException(..)) import Data.IORef import qualified Network.HTTP.Types as H+import Network.Socket (SockAddr) import qualified System.TimeManager as T import UnliftIO.Exception (SomeException(..)) import qualified UnliftIO.Exception as E@@ -33,6 +33,8 @@   , isPushable     :: IO Bool   , insertStream   :: StreamId -> a -> IO ()   , makePushStream :: a -> PushPromise -> IO (StreamId, StreamId, a)+  , mySockAddr     :: SockAddr+  , peerSockAddr   :: SockAddr   }  fromContext :: Context -> WorkerConf Stream@@ -53,6 +55,8 @@         newstrm <- newPushStream sid ws         let pid = streamNumber pstrm         return (pid, sid, newstrm)+  , mySockAddr = mySockAddr+  , peerSockAddr = peerSockAddr   }  ----------------------------------------------------------------@@ -137,12 +141,11 @@             atomically $ writeTBQueue tbq (StreamingBuilder b)             T.resume th           flush  = atomically $ writeTBQueue tbq StreamingFlush-      strmbdy push flush-      atomically $ writeTBQueue tbq StreamingFinished-      -- Remove the thread's ID from the manager's queue, to ensure the that the-      -- manager will not terminate it before we are done. (The thread ID was-      -- added implicitly when the worker was spawned by the manager).-      deleteMyId mgr+          finished = atomically $ writeTBQueue tbq $ StreamingFinished (decCounter mgr)+      incCounter mgr+      strmbdy push flush `E.finally` finished+  OutBodyStreamingUnmask _ ->+    error "response: server does not support OutBodyStreamingUnmask"   where     (_,reqvt) = inpObjHeaders req @@ -162,13 +165,13 @@             setStreamInfo sinfo strm             T.resume th             T.tickle th-            let aux = Aux th+            let aux = Aux th mySockAddr peerSockAddr             server (Request req') aux $ response wc mgr th tcont strm (Request req')         cont1 <- case ex of             Right () -> return True             Left e@(SomeException _)               -- killed by the local worker manager-              | Just ThreadKilled    <- E.fromException e -> return False+              | Just KilledByHttp2ThreadManager{} <- E.fromException e -> return False               -- killed by the local timeout manager               | Just T.TimeoutThread <- E.fromException e -> do                   cleanup sinfo

@@ -1,6 +1,6 @@ cabal-version:      >=1.10 name:               http2-version:            4.1.4+version:            4.2.0 license:            BSD3 license-file:       LICENSE maintainer:         Kazu Yamamoto <kazu@iij.ad.jp>@@ -288,12 +288,12 @@     build-tool-depends: hspec-discover:hspec-discover     hs-source-dirs:     test     other-modules:-        HTTP2.ClientSpec         HPACK.DecodeSpec         HPACK.EncodeSpec         HPACK.HeaderBlock         HPACK.HuffmanSpec         HPACK.IntegerSpec+        HTTP2.ClientSpec         HTTP2.FrameSpec         HTTP2.ServerSpec @@ -311,6 +311,7 @@         http2,         network,         network-run >=0.1.0,+        random,         typed-process  test-suite spec2

@@ -1,9 +1,9 @@ module Main where -import Control.Monad (when) import Data.Aeson import Data.Aeson.Encode.Pretty (encodePretty) import qualified Data.ByteString.Lazy.Char8 as BL+import Data.Maybe (fromJust) import System.Environment (getArgs) import System.Exit (exitFailure) import System.IO (hPutStrLn, stderr)@@ -14,11 +14,12 @@ main :: IO () main = do     args <- getArgs-    when (length args /= 3) $ do+    (arg1,arg2,desc) <- case args of+      [a,b,c] -> return (a,b,c)+      _     -> do         hPutStrLn stderr "hpack-encode on/off naive|linear <desc>"         exitFailure-    let [arg1,arg2,desc] = args-        huffman+    let huffman           | arg1 == "on" = True           | otherwise = False         algo@@ -31,7 +32,7 @@ hpackEncode :: EncodeStrategy -> String -> IO () hpackEncode stgy desc = do     bs <- BL.getContents-    let Just tc = decode bs :: Maybe Test+    let tc = fromJust (decode bs :: Maybe Test)     hexs <- run False stgy tc     let cs = cases tc         cs' = zipWith update cs hexs

@@ -9,6 +9,7 @@ import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy.Char8 as BL import Data.List+import Data.Maybe (fromJust) import System.Directory import System.FilePath import Control.Monad@@ -75,7 +76,7 @@ getHeaderSize :: FilePath -> IO Int getHeaderSize file = do     bs <- BL.readFile file-    let Just tc = decode bs :: Maybe Test+    let tc = fromJust (decode bs :: Maybe Test)     let len = sum $ map toT $ cases tc     return len   where@@ -84,14 +85,14 @@ getHeaderLen :: FilePath -> IO Int getHeaderLen file = do     bs <- BL.readFile file-    let Just tc = decode bs :: Maybe Test+    let tc = fromJust (decode bs :: Maybe Test)     let len = length $ cases tc     return len  getWireSize :: FilePath -> IO Int getWireSize file = do     bs <- BL.readFile file-    let Just tc = decode bs :: Maybe Test+    let tc = fromJust (decode bs :: Maybe Test)     let len = sum $ map toT $ cases tc     return len   where

@@ -10,13 +10,17 @@ import qualified Data.ByteString.Char8 as C8 import Network.HTTP.Types import Network.Run.TCP+import System.IO.Unsafe (unsafePerformIO)+import System.Random import Test.Hspec  import Network.HTTP2.Client import qualified Network.HTTP2.Server as S  port :: String-port = "8080"+port = show $ unsafePerformIO (randomPort <$> getStdGen)+  where+    randomPort = fst . randomR (43124 :: Int, 44320)  host :: String host = "127.0.0.1"

@@ -1,5 +1,6 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RankNTypes #-}  module HTTP2.ServerSpec where @@ -9,17 +10,19 @@ import Control.Monad import Crypto.Hash (Context, SHA1) -- cryptonite import qualified Crypto.Hash as CH+import Data.ByteString (ByteString) import qualified Data.ByteString as B import Data.ByteString.Builder (byteString, Builder)-import Data.ByteString.Char8 import qualified Data.ByteString.Char8 as C8 import Data.IORef import Network.HTTP.Types import Network.Run.TCP import Network.Socket import Network.Socket.ByteString-import Test.Hspec import System.IO+import System.IO.Unsafe (unsafePerformIO)+import System.Random+import Test.Hspec  import Network.HPACK import Network.HPACK.Token@@ -28,7 +31,9 @@ import Network.HTTP2.Frame  port :: String-port = "8080"+port = show $ unsafePerformIO (randomPort <$> getStdGen)+  where+    randomPort = fst . randomR (43124 :: Int, 44320)  host :: String host = "127.0.0.1"@@ -158,9 +163,19 @@     runHTTP2Client s = E.bracket (allocConfig s 4096)                                  freeSimpleConfig                                  (\conf -> C.run cliconf conf client)-    client sendRequest = mapConcurrently_ ($ sendRequest) clients-    clients = [client0,client1,client2,client3,client3',client3'',client4,client5] +    client :: C.Client ()+    client sendRequest = foldr1 concurrently_ $ [+          client0   sendRequest+        , client1   sendRequest+        , client2   sendRequest+        , client3   sendRequest+        , client3'  sendRequest+        , client3'' sendRequest+        , client4   sendRequest+        , client5   sendRequest+        ]+ -- delay sending preface to be able to test if it is always sent first allocSlowPrefaceConfig :: Socket -> BufferSize -> IO Config allocSlowPrefaceConfig s size = do@@ -169,7 +184,7 @@   where     slowPrefaceSend :: (ByteString -> IO ()) -> ByteString -> IO ()     slowPrefaceSend orig chunk = do-      when (C8.pack "PRI" `isPrefixOf` chunk) $ do+      when (C8.pack "PRI" `C8.isPrefixOf` chunk) $ do         threadDelay 10000       orig chunk 

@@ -1,10 +1,10 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}  module Main where  import Control.Concurrent.Async import qualified Control.Exception as E-import Control.Monad import qualified Data.ByteString.Char8 as C8 import Network.HTTP.Types import Network.Run.TCP (runTCPClient) -- network-run@@ -19,16 +19,18 @@ main :: IO () main = do     args <- getArgs-    when (length args /= 2) $ do-        putStrLn "client <addr> <port>"-        exitFailure-    let [host,port] = args+    (host,port) <- case args of+      [h,p] -> return (h,p)+      _     -> do+          putStrLn "client <addr> <port>"+          exitFailure     runTCPClient serverName port $ runHTTP2Client host   where     cliconf host = ClientConfig "http" (C8.pack host) 20     runHTTP2Client host s = E.bracket (allocSimpleConfig s 4096)                                       freeSimpleConfig                                       (\conf -> run (cliconf host) conf client)+    client :: Client ()     client sendRequest = do         let req0 = requestNoBody methodGet "/" []             client0 = sendRequest req0 $ \rsp -> do

@@ -23,10 +23,11 @@ main :: IO () main = do     args <- getArgs-    when (length args /= 2) $ do-        putStrLn "server <addr> <port>"-        exitFailure-    let [host,port] = args+    (host,port) <- case args of+      [h,p] -> return (h,p)+      _     -> do+          putStrLn "server <addr> <port>"+          exitFailure     runTCPServer (Just host) port runHTTP2Server   where     runHTTP2Server s = E.bracket (allocSimpleConfig s 4096)