diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/Network/HTTP2/Arch/Config.hs b/Network/HTTP2/Arch/Config.hs --- a/Network/HTTP2/Arch/Config.hs +++ b/Network/HTTP2/Arch/Config.hs @@ -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 diff --git a/Network/HTTP2/Arch/Context.hs b/Network/HTTP2/Arch/Context.hs --- a/Network/HTTP2/Arch/Context.hs +++ b/Network/HTTP2/Arch/Context.hs @@ -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 diff --git a/Network/HTTP2/Arch/Manager.hs b/Network/HTTP2/Arch/Manager.hs --- a/Network/HTTP2/Arch/Manager.hs +++ b/Network/HTTP2/Arch/Manager.hs @@ -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) diff --git a/Network/HTTP2/Arch/Receiver.hs b/Network/HTTP2/Arch/Receiver.hs --- a/Network/HTTP2/Arch/Receiver.hs +++ b/Network/HTTP2/Arch/Receiver.hs @@ -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 diff --git a/Network/HTTP2/Arch/Sender.hs b/Network/HTTP2/Arch/Sender.hs --- a/Network/HTTP2/Arch/Sender.hs +++ b/Network/HTTP2/Arch/Sender.hs @@ -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 diff --git a/Network/HTTP2/Arch/Stream.hs b/Network/HTTP2/Arch/Stream.hs --- a/Network/HTTP2/Arch/Stream.hs +++ b/Network/HTTP2/Arch/Stream.hs @@ -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' diff --git a/Network/HTTP2/Arch/Types.hs b/Network/HTTP2/Arch/Types.hs --- a/Network/HTTP2/Arch/Types.hs +++ b/Network/HTTP2/Arch/Types.hs @@ -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 diff --git a/Network/HTTP2/Client.hs b/Network/HTTP2/Client.hs --- a/Network/HTTP2/Client.hs +++ b/Network/HTTP2/Client.hs @@ -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 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 @@ -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 diff --git a/Network/HTTP2/Client/Types.hs b/Network/HTTP2/Client/Types.hs --- a/Network/HTTP2/Client/Types.hs +++ b/Network/HTTP2/Client/Types.hs @@ -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) diff --git a/Network/HTTP2/Internal.hs b/Network/HTTP2/Internal.hs --- a/Network/HTTP2/Internal.hs +++ b/Network/HTTP2/Internal.hs @@ -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 diff --git a/Network/HTTP2/Server.hs b/Network/HTTP2/Server.hs --- a/Network/HTTP2/Server.hs +++ b/Network/HTTP2/Server.hs @@ -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 diff --git a/Network/HTTP2/Server/Run.hs b/Network/HTTP2/Server/Run.hs --- a/Network/HTTP2/Server/Run.hs +++ b/Network/HTTP2/Server/Run.hs @@ -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 diff --git a/Network/HTTP2/Server/Types.hs b/Network/HTTP2/Server/Types.hs --- a/Network/HTTP2/Server/Types.hs +++ b/Network/HTTP2/Server/Types.hs @@ -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 } 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 @@ -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 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: 4.1.4 +version: 4.2.0 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -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 diff --git a/test-hpack/hpack-encode.hs b/test-hpack/hpack-encode.hs --- a/test-hpack/hpack-encode.hs +++ b/test-hpack/hpack-encode.hs @@ -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 " 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 diff --git a/test-hpack/hpack-stat.hs b/test-hpack/hpack-stat.hs --- a/test-hpack/hpack-stat.hs +++ b/test-hpack/hpack-stat.hs @@ -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 diff --git a/test/HTTP2/ClientSpec.hs b/test/HTTP2/ClientSpec.hs --- a/test/HTTP2/ClientSpec.hs +++ b/test/HTTP2/ClientSpec.hs @@ -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" diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -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 diff --git a/util/client.hs b/util/client.hs --- a/util/client.hs +++ b/util/client.hs @@ -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 " - exitFailure - let [host,port] = args + (host,port) <- case args of + [h,p] -> return (h,p) + _ -> do + putStrLn "client " + 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 diff --git a/util/server.hs b/util/server.hs --- a/util/server.hs +++ b/util/server.hs @@ -23,10 +23,11 @@ main :: IO () main = do args <- getArgs - when (length args /= 2) $ do - putStrLn "server " - exitFailure - let [host,port] = args + (host,port) <- case args of + [h,p] -> return (h,p) + _ -> do + putStrLn "server " + exitFailure runTCPServer (Just host) port runHTTP2Server where runHTTP2Server s = E.bracket (allocSimpleConfig s 4096)