diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -2,12 +2,13 @@ {-# LANGUAGE RecordWildCards #-} module Main where -import Control.Monad (forever, when, void) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.Async (async, waitAnyCancel) +import Control.Monad (forever, when, void) import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as ByteString import Data.Default.Class (def) +import Data.Maybe (fromMaybe) import Data.Time.Clock (diffUTCTime, getCurrentTime) import qualified Network.HTTP2 as HTTP2 import qualified Network.TLS as TLS @@ -32,28 +33,33 @@ data ServerPushSwitch = PushEnabled | PushDisabled deriving Show +data Verbosity = Verbose | NonVerbose + deriving Show + data QueryArgs = QueryArgs { - _host :: !HostName - , _port :: !PortNumber - , _verb :: !Verb - , _path :: !Path - , _extraHeaders :: ![(ByteString, ByteString)] - , _postData :: !(Maybe PostData) - , _interPingDelay :: !Int - , _pingTimeout :: !Int - , _interFlowControlUpdates :: !Int - , _settingsMaxConcurrency :: !Int - , _settingsAllowServerPush :: !ServerPushSwitch + _host :: !HostName + , _port :: !PortNumber + , _verb :: !Verb + , _path :: !Path + , _extraHeaders :: ![(ByteString, ByteString)] + , _postData :: !(Maybe PostData) + , _interPingDelay :: !Int + , _pingTimeout :: !Int + , _interFlowControlUpdates :: !Int + , _settingsMaxConcurrency :: !Int + , _settingsAllowServerPush :: !ServerPushSwitch , _settingsMaxFrameSize :: !Int , _settingsMaxHeaderBlockSize :: !Int , _settingsInitialWindowSize :: !Int , _initialWindowKick :: !Int , _concurrentQueriesCount :: !Int , _numberQueries :: !Int - , _finalDelay :: !Int - , _finalMessage :: !ByteString - , _encoderBufsize :: !Int - , _decoderBufsize :: !Int + , _finalDelay :: !Int + , _finalMessage :: !ByteString + , _encoderBufsize :: !Int + , _decoderBufsize :: !Int + , _downloadPrefix :: !FilePath + , _verboseDebug :: !Verbosity } deriving Show clientArgs :: Parser QueryArgs @@ -80,6 +86,8 @@ <*> kthxByeMessage <*> encoderBufSize <*> decoderBufSize + <*> downloadPrefix + <*> verboseDebug where bstrOption = fmap ByteString.pack . strOption milliseconds what base = fmap (*1000) $ option auto (long what <> value base) @@ -102,6 +110,8 @@ kthxByeMessage = bstrOption (long "exit-greeting" <> value "kthxbye (>;_;<)") encoderBufSize = option auto (long "hpack-encoder-buffer-size" <> value 4096) decoderBufSize = option auto (long "hpack-decoder-buffer-size" <> value 4096) + downloadPrefix = strOption (long "push-files-prefix" <> value ":stdout-pp") + verboseDebug = flag NonVerbose Verbose (long "verbose") main :: IO () main = execParser opts >>= client @@ -136,9 +146,11 @@ , (":authority", ByteString.pack _host) ] <> _extraHeaders - let ppHandler _ stream ppHdrs streamFlowControl _ = void $ forkIO $ do - timePrint ("push stream started" :: String, ppHdrs) - waitStream stream streamFlowControl >>= timePrint . fromStreamResult + let ppHandler n idx _ stream ppHdrs streamFlowControl _ = void $ forkIO $ do + let pushpath = fromMaybe "unspecified-path" (lookup ":path" ppHdrs) + timePrint ("push stream started" :: String, pushpath) + ret <- fromStreamResult <$> waitStream stream streamFlowControl + either (\e -> timePrint e) (dump PushPromiseFile pushpath n idx _downloadPrefix) ret timePrint ("push stream ended" :: String) let conf = [ (HTTP2.SettingsMaxFrameSize, _settingsMaxFrameSize) @@ -150,7 +162,32 @@ ] timePrint conf - conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf + frameConn <- newHttp2FrameConnection _host _port tlsParams + let wrappedFrameConn = frameConn { + _makeFrameClientStream = \sid -> + let frameClient = (_makeFrameClientStream frameConn) sid + in frameClient { + _sendFrames = \xs -> do + print $ (">>> "::String, _getStreamId frameClient, map snd xs) + _sendFrames frameClient xs + } + , _serverStream = + let + currentServerStrean = _serverStream frameConn + in + currentServerStrean { + _nextHeaderAndFrame = do + hdrFrame@(hdr,_) <- _nextHeaderAndFrame currentServerStrean + print ("<<< "::String, HTTP2.streamId hdr, hdrFrame) + return hdrFrame + } + } + conn <- case _verboseDebug of + Verbose -> + wrapFrameClient wrappedFrameConn _encoderBufsize _decoderBufsize conf + NonVerbose -> + wrapFrameClient frameConn _encoderBufsize _decoderBufsize conf + linkHttp2Client conn _addCredit (_incomingFlowControl conn) _initialWindowKick _ <- forkIO $ forever $ do updated <- _updateWindow $ _incomingFlowControl conn @@ -165,15 +202,17 @@ let go 0 idx = timePrint $ "done worker: " <> show idx go n idx = do _ <- (withHttp2Stream conn $ \stream -> - let initStream = headers stream headersPairs headersFlags + let initStream = do + _ <- async $ onPushPromise stream (ppHandler n idx) + headers stream headersPairs headersFlags handler streamINFlowControl streamOUTFlowControl = do - _ <- async $ onPushPromise stream ppHandler timePrint $ "stream started " <> show (idx, n) _ <- async $ dataPostFunction conn (_outgoingFlowControl conn) stream streamOUTFlowControl - waitStream stream streamINFlowControl >>= timePrint . fromStreamResult + ret <- fromStreamResult <$> waitStream stream streamINFlowControl + either (\e -> timePrint e) (dump MainFile _path n idx _downloadPrefix) ret timePrint $ "stream ended " <> show (idx, n) in StreamDefinition initStream handler) go (n - 1) idx @@ -198,7 +237,37 @@ , TLS.clientDebug = def } +data DumpType = MainFile | PushPromiseFile + +dump :: DumpType -> Path -> Int -> Int -> FilePath -> StreamResponse -> IO () +dump MainFile _ _ _ ":none" (hdrs, _) = do + timePrint hdrs +dump MainFile _ _ _ ":stdout" (hdrs, body) = do + timePrint hdrs + ByteString.putStrLn body +dump PushPromiseFile _ _ _ ":stdout" (hdrs, _) = do + timePrint hdrs +dump MainFile _ _ _ ":stdout-pp" (hdrs, body) = do + timePrint hdrs + ByteString.putStrLn body +dump PushPromiseFile _ _ _ ":stdout-pp" (hdrs, body) = do + timePrint hdrs + ByteString.putStrLn body +dump _ querystring nquery nthread prefix (hdrs, body) = do + timePrint hdrs + ByteString.writeFile filepath body + where + filepath = mconcat [ prefix + , "/" + , show nquery + , "." + , show nthread + , ByteString.unpack $ cleanPath querystring + ] + cleanPath = ByteString.takeWhile (/= '?') + . ByteString.map (\c -> if c == '/' then '_' else c) + timePrint :: Show a => a -> IO () timePrint x = do tst <- getCurrentTime - print (tst, x) + ByteString.hPutStrLn stderr $ ByteString.pack $ show (tst, x) diff --git a/http2-client.cabal b/http2-client.cabal --- a/http2-client.cabal +++ b/http2-client.cabal @@ -1,5 +1,5 @@ name: http2-client -version: 0.3.0.0 +version: 0.3.0.1 synopsis: A native HTTP2 client library. description: Please read the README.md at the homepage. homepage: https://github.com/lucasdicioccio/http2-client diff --git a/src/Network/HTTP2/Client.hs b/src/Network/HTTP2/Client.hs --- a/src/Network/HTTP2/Client.hs +++ b/src/Network/HTTP2/Client.hs @@ -24,8 +24,11 @@ -- * Flow control , IncomingFlowControl(..) , OutgoingFlowControl(..) + -- * Exceptions + , linkHttp2Client -- * Misc. , FlagSetter + , Http2ClientAsyncs(..) , wrapFrameClient , _gtfo -- * Convenience re-exports @@ -35,11 +38,11 @@ ) where import Control.Exception (bracket, throw) -import Control.Concurrent.Async (race) +import Control.Concurrent.Async (Async, async, race, link) import Control.Concurrent.MVar (newMVar, takeMVar, putMVar) -import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent (threadDelay) import Control.Concurrent.Chan (Chan, newChan, dupChan, readChan, writeChan) -import Control.Monad (forever, void, when, forM_) +import Control.Monad (forever, when, forM_) import Data.ByteString (ByteString) import qualified Data.ByteString as ByteString import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef) @@ -165,8 +168,45 @@ -- connection. , _paylodSplitter :: IO PayloadSplitter -- ^ Returns a function to split a payload. + , _asyncs :: !Http2ClientAsyncs + -- ^ Asynchronous operations threads. } +-- | Set of Async threads running an Http2Client. +-- +-- If you add other asyncs here, you should also modify +-- 'linkHttp2Client'. +data Http2ClientAsyncs = Http2ClientAsyncs { + _waitSettingsAsync :: Async (FrameHeader, FramePayload) + -- ^ Async waiting for the initial settings ACK. + , _creditFlowsAsync :: Async () + -- ^ Async responsible for crediting the connection incoming flow control. + -- See 'creditDataFramesLoop'. + , _HPACKAsync :: Async () + -- ^ Async responsible for serializing every HEADERS/PUSH_PROMISE/CONTINUATION. + -- See 'incomingHPACKFramesLoop'. + , _controlFramesAsync :: Async () + -- ^ Async responsible for handling control frames (e.g., PING). + -- See 'incomingControlFramesLoop'. + , _incomingFramesAsync :: Async () + -- ^ Async responsible for ingesting all frames, increasing the + -- maximum-received streamID and starting the frame dispatch. See + -- 'incomingFramesLoop'. + } + +-- | Links all asyncs threads of a client to the current thread. +-- +-- See 'link'. +linkHttp2Client :: Http2Client -> IO () +linkHttp2Client client = + let Http2ClientAsyncs{..} = _asyncs client + in do + link _waitSettingsAsync + link _creditFlowsAsync + link _HPACKAsync + link _controlFramesAsync + link _incomingFramesAsync + -- | Couples client and server settings together. data ConnectionSettings = ConnectionSettings { _clientSettings :: !Settings @@ -329,12 +369,12 @@ -- Initial thread receiving server frames. maxReceivedStreamId <- newIORef 0 serverFrames <- newChan - _ <- forkIO $ incomingFramesLoop conn serverFrames maxReceivedStreamId + aIncoming <- async $ incomingFramesLoop conn serverFrames maxReceivedStreamId -- Thread handling control frames. settings <- newIORef defaultConnectionSettings controlFrames <- dupChan serverFrames - _ <- forkIO $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings + aControl <- async $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings -- Thread handling push-promises and headers frames serializing the buffers. serverStreamFrames <- dupChan serverFrames @@ -350,13 +390,13 @@ serverPushPromises <- newChan - _ <- forkIO $ incomingHPACKFramesLoop serverStreamFrames - serverHeaders - serverPushPromises - hpackDecoder + aHPACK <- async $ incomingHPACKFramesLoop serverStreamFrames + serverHeaders + serverPushPromises + hpackDecoder dataFrames <- dupChan serverFrames - _ <- forkIO $ creditDataFramesLoop _incomingFlowControl dataFrames + aCredit <- async $ creditDataFramesLoop _incomingFlowControl dataFrames conccurentStreams <- newIORef 0 let _startStream getWork = do @@ -403,8 +443,10 @@ let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings - _ <- forkIO . void =<< _settings initSettings + aSettings <- async =<< _settings initSettings + let _asyncs = Http2ClientAsyncs aSettings aCredit aHPACK aControl aIncoming + return $ Http2Client{..} initializeStream @@ -539,7 +581,8 @@ | OpenPushPromise !StreamId !StreamId type FramesChan e = Chan (FrameHeader, Either e FramePayload) -type HeadersChan = Chan (FrameHeader, StreamId, Either ErrorCode HeaderList) +type HeadersChanContent = (FrameHeader, StreamId, Either ErrorCode HeaderList) +type HeadersChan = Chan HeadersChanContent type PushPromisesChanContent e = (StreamId, FramesChan e, HeadersChan, StreamId, HeaderList) type PushPromisesChan e = Chan (PushPromisesChanContent e) @@ -606,12 +649,15 @@ -> Http2FrameClientStream -> IO IncomingFlowControl newIncomingFlowControl settings stream = do + let getBase = if _getStreamId stream == 0 + then return HTTP2.defaultInitialWindowSize + else initialWindowSize . _clientSettings <$> readIORef settings creditAdded <- newIORef 0 creditConsumed <- newIORef 0 let _addCredit n = atomicModifyIORef' creditAdded (\c -> (c + n, ())) let _consumeCredit n = do conso <- atomicModifyIORef' creditConsumed (\c -> (c + n, c + n)) - base <- initialWindowSize . _clientSettings <$> readIORef settings + base <- getBase extra <- readIORef creditAdded return $ base + extra - conso let _updateWindow = do @@ -638,10 +684,13 @@ -> IO OutgoingFlowControl newOutgoingFlowControl settings sid frames = do credit <- newIORef 0 + let getBase = if sid == 0 + then return HTTP2.defaultInitialWindowSize + else initialWindowSize . _serverSettings <$> readIORef settings let receive n = atomicModifyIORef' credit (\c -> (c + n, ())) let withdraw 0 = return 0 withdraw n = do - base <- initialWindowSize . _serverSettings <$> readIORef settings + base <- getBase got <- atomicModifyIORef' credit (\c -> if base + c >= n then (c - n, n) @@ -789,21 +838,24 @@ return () waitFrameWithStreamId - :: Exception e => - StreamId -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload) + :: Exception e + => StreamId + -> FramesChan e + -> IO (FrameHeader, FramePayload) waitFrameWithStreamId sid = waitFrame (\h _ -> streamId h == sid) waitFrameWithTypeId :: (Exception e) => [FrameTypeId] - -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload) + -> FramesChan e + -> IO (FrameHeader, FramePayload) waitFrameWithTypeId tids = waitFrame (\_ p -> HTTP2.framePayloadToFrameTypeId p `elem` tids) waitFrameWithTypeIdForStreamId :: (Exception e) => StreamId -> [FrameTypeId] - -> Chan (FrameHeader, Either e FramePayload) + -> FramesChan e -> IO (FrameHeader, FramePayload) waitFrameWithTypeIdForStreamId sid tids = waitFrame (\h p -> streamId h == sid && HTTP2.framePayloadToFrameTypeId p `elem` tids) @@ -811,7 +863,7 @@ waitFrame :: Exception e => (FrameHeader -> FramePayload -> Bool) - -> Chan (FrameHeader, Either e FramePayload) + -> FramesChan e -> IO (FrameHeader, FramePayload) waitFrame test chan = loop @@ -833,15 +885,15 @@ waitHeadersWithStreamId :: StreamId - -> Chan (FrameHeader, StreamId, t) - -> IO (FrameHeader, StreamId, t) + -> HeadersChan + -> IO HeadersChanContent waitHeadersWithStreamId sid = waitHeaders (\_ s _ -> s == sid) waitHeaders - :: (FrameHeader -> StreamId -> t -> Bool) - -> Chan (FrameHeader, StreamId, t) - -> IO (FrameHeader, StreamId, t) + :: (FrameHeader -> StreamId -> Either ErrorCode HeaderList -> Bool) + -> HeadersChan + -> IO HeadersChanContent waitHeaders test chan = loop where