diff --git a/cbits/tlsinc.c b/cbits/tlsinc.c --- a/cbits/tlsinc.c +++ b/cbits/tlsinc.c @@ -456,7 +456,7 @@ // Now I set a few options.... /*SSL_CTX_set_verify(c->sslContext, NULL );*/ const long flags = - SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_TLSv1 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION; + SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION; // const long flags = // SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION; SSL_CTX_set_options(c->sslContext, flags); @@ -469,9 +469,9 @@ // The only cipher supported by HTTP/2 ... sort of. result = SSL_CTX_set_cipher_list(c->sslContext, - "ECDHE-RSA-AES256-GCM-SHA384" // <-- Good for Chrome and firefox - ":ECDHE-RSA-AES128-GCM-SHA256" // -- ibidem - // ---- These suites will fail HTTP/2 + "ECDHE-RSA-AES128-GCM-SHA256" // -- ibidem + ":ECDHE-RSA-AES256-GCM-SHA384" // <-- Good for Chrome and firefox + // ---- These suites will fail HTTP/2 ":ECDHE-RSA-AES128-SHA256" // -- Not good for HTTP/2 ":ECDHE-RSA-AES-128-CBC-SHA256" ":ECDHE-ECDSA-AES256-SHA384" diff --git a/hs-src/SecondTransfer.hs b/hs-src/SecondTransfer.hs --- a/hs-src/SecondTransfer.hs +++ b/hs-src/SecondTransfer.hs @@ -69,13 +69,15 @@ helloWorldWorker :: CoherentWorker -helloWorldWorker request = return ( - [ - (":status", "200") - ], - [], -- No pushed streams - saysHello - ) +helloWorldWorker (_request_headers, _maybe_post_data) = do + dropIncomingData _maybe_post_data + return ( + [ + (":status", "200") + ], + [], -- No pushed streams + saysHello + ) -- For this program to work, it should be run from the top of @@ -156,6 +158,7 @@ -- HTTP/2 server in a snap. ,tlsServeWithALPN ,tlsServeWithALPNAndFinishOnRequest + ,dropIncomingData ,TLSLayerGenericProblem(..) ,FinishRequest(..) @@ -175,3 +178,4 @@ import SecondTransfer.MainLoop import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.Types +import SecondTransfer.Utils.DevNull (dropIncomingData) diff --git a/hs-src/SecondTransfer/Exception.hs b/hs-src/SecondTransfer/Exception.hs --- a/hs-src/SecondTransfer/Exception.hs +++ b/hs-src/SecondTransfer/Exception.hs @@ -15,6 +15,9 @@ ,IOProblem(..) ,GenericIOProblem(..) ,StreamCancelledException(..) + + -- * Internal exceptions + ,HTTP2ProtocolException(..) ) where import Control.Exception @@ -39,6 +42,14 @@ HTTP2SessionException a <- fromException x cast a +-- | Concrete exception. Used internally to signal that the client violated +-- the protocol. Clients of the library shall never see this exception. +data HTTP2ProtocolException = HTTP2ProtocolException + deriving (Typeable, Show) + +instance Exception HTTP2ProtocolException where + toException = convertHTTP2SessionExceptionToException + fromException = getHTTP2SessionExceptionFromException -- | Abstract exception. Thrown when encoding/decoding of a frame fails diff --git a/hs-src/SecondTransfer/Http1/Parse.hs b/hs-src/SecondTransfer/Http1/Parse.hs --- a/hs-src/SecondTransfer/Http1/Parse.hs +++ b/hs-src/SecondTransfer/Http1/Parse.hs @@ -32,9 +32,11 @@ import qualified Data.ByteString.Lazy as Lb import Data.Char (toLower) import Data.Maybe (isJust) +#ifndef IMPLICIT_MONOID import Data.Monoid (mappend, mempty, mconcat) +#endif import qualified Data.Attoparsec.ByteString as Ap import Data.Foldable (find) @@ -46,7 +48,7 @@ import SecondTransfer.MainLoop.CoherentWorker (Headers) import SecondTransfer.Utils (subByteString) - +import Prelude data IncrementalHttp1Parser = IncrementalHttp1Parser { diff --git a/hs-src/SecondTransfer/Http1/Session.cpphs b/hs-src/SecondTransfer/Http1/Session.cpphs deleted file mode 100644 --- a/hs-src/SecondTransfer/Http1/Session.cpphs +++ /dev/null @@ -1,108 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -{-# OPTIONS_HADDOCK hide #-} -module SecondTransfer.Http1.Session( - http11Attendant - ) where - - --- import qualified Control.Lens as L --- import Control.Lens ( (^.) ) -import Control.Exception (catch) -import Control.Concurrent (forkIO) - -import qualified Data.ByteString as B --- import qualified Data.ByteString.Lazy as LB --- import Data.ByteString.Char8 (unpack) --- import qualified Data.ByteString.Builder as Bu -import Data.Conduit -import Data.Conduit.List (consume) --- import Data.Monoid (mconcat, mappend) - -import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker,) -import SecondTransfer.MainLoop.PushPullType (Attendant) -import SecondTransfer.Sessions.Internal (SessionsContext, acquireNewSessionTag) - --- Logging utilities -import System.Log.Logger - -import SecondTransfer.Http1.Parse -import SecondTransfer.Exception (IOProblem) - --- import Debug.Trace (traceShow) - - --- | Session attendant that speaks HTTP/1.1 --- -http11Attendant :: SessionsContext -> CoherentWorker -> Attendant -http11Attendant sessions_context coherent_worker - push_action pull_action close_action - = do - new_session_tag <- acquireNewSessionTag sessions_context - infoM "Session.Session_HTTP11" $ "Starting new session with tag: " ++(show new_session_tag) - forkIO $ go new_session_tag (Just "") - return () - where - go :: Int -> Maybe B.ByteString -> IO () - go session_tag (Just leftovers) = do - infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++(show session_tag) - maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag - go session_tag maybe_leftovers - - go _ Nothing = - return () - - add_data :: IncrementalHttp1Parser -> B.ByteString -> Int -> IO (Maybe B.ByteString) - add_data parser bytes session_tag = do - let - completion = addBytes parser bytes - -- completion = addBytes parser $ traceShow ("At session " ++ (show session_tag) ++ " Received: " ++ (unpack bytes) ) bytes - case completion of - - MustContinue_H1PC new_parser -> do - -- print "MustContinue_H1PC" - catch - (do - new_bytes <- pull_action - r <- add_data new_parser new_bytes session_tag - return r - ) - ( (\ _e -> do - -- This is a pretty harmless condition that happens - -- often when the remote peer closes the connection - debugM "Session.HTTP1" "Could not receive data" - close_action - return Nothing - ) :: IOProblem -> IO (Maybe B.ByteString) ) - - - - OnlyHeaders_H1PC headers leftovers -> do - -- print "OnlyHeaders_H1PC" - -- Ready for action... - -- ATTENTION: Not use for pushed streams here.... - -- We must decide what to do if the user return those - -- anyway. - (response_headers, _, data_and_conclusion) <- coherent_worker (headers, Nothing) - (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume - let - response_text = - serializeHTTPResponse response_headers fragments - - catch - (do - push_action response_text - return $ Just leftovers - ) - ((\ _e -> do - debugM "Session.HTTP1" "Session abandoned" - close_action - return Nothing - ) :: IOProblem -> IO (Maybe B.ByteString) ) - - HeadersAndBody_H1PC _headers _stopcondition _recv_leftovers -> do - -- print "HeadersAndBody_H1PC" - -- Let's see if I can go through the basic movements first, then through - -- more complicated things. - -- TODO: Implement posts and other requests with bodies.... - close_action - error "NotImplemented requests with bodies" diff --git a/hs-src/SecondTransfer/Http1/Session.hs b/hs-src/SecondTransfer/Http1/Session.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http1/Session.hs @@ -0,0 +1,136 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_HADDOCK hide #-} +module SecondTransfer.Http1.Session( + http11Attendant + ) where + + +import Control.Lens +import Control.Exception (catch) +import Control.Concurrent (forkIO) + +import qualified Data.ByteString as B +-- import qualified Data.ByteString.Lazy as LB +-- import Data.ByteString.Char8 (unpack) +-- import qualified Data.ByteString.Builder as Bu +import Data.Conduit +import Data.Conduit.List (consume) +-- import Data.Monoid (mconcat, mappend) + +import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker,Headers) +import SecondTransfer.MainLoop.PushPullType (Attendant) +import SecondTransfer.Sessions.Internal (SessionsContext, acquireNewSessionTag) + +-- Logging utilities +import System.Log.Logger + +import SecondTransfer.Http1.Parse +import SecondTransfer.Exception (IOProblem) +import SecondTransfer.Sessions.Config +import SecondTransfer.Sessions.Internal (sessionsConfig) +import qualified SecondTransfer.Utils.HTTPHeaders as He + +-- import Debug.Trace (traceShow) + + +-- | Session attendant that speaks HTTP/1.1 +-- +http11Attendant :: SessionsContext -> CoherentWorker -> Attendant +http11Attendant sessions_context coherent_worker + push_action pull_action close_action + = do + new_session_tag <- acquireNewSessionTag sessions_context + infoM "Session.Session_HTTP11" $ "Starting new session with tag: " ++(show new_session_tag) + forkIO $ go new_session_tag (Just "") + return () + where + go :: Int -> Maybe B.ByteString -> IO () + go session_tag (Just leftovers) = do + infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++(show session_tag) + maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag + go session_tag maybe_leftovers + + go _ Nothing = + return () + + add_data :: IncrementalHttp1Parser -> B.ByteString -> Int -> IO (Maybe B.ByteString) + add_data parser bytes session_tag = do + let + completion = addBytes parser bytes + -- completion = addBytes parser $ traceShow ("At session " ++ (show session_tag) ++ " Received: " ++ (unpack bytes) ) bytes + case completion of + + MustContinue_H1PC new_parser -> do + -- print "MustContinue_H1PC" + catch + (do + new_bytes <- pull_action + r <- add_data new_parser new_bytes session_tag + return r + ) + ( (\ _e -> do + -- This is a pretty harmless condition that happens + -- often when the remote peer closes the connection + debugM "Session.HTTP1" "Could not receive data" + close_action + return Nothing + ) :: IOProblem -> IO (Maybe B.ByteString) ) + + + OnlyHeaders_H1PC headers leftovers -> do + -- print "OnlyHeaders_H1PC" + -- Ready for action... + -- ATTENTION: Not use for pushed streams here.... + -- We must decide what to do if the user return those + -- anyway. + let + modified_headers = addExtraHeaders sessions_context headers + (response_headers, _, data_and_conclusion) <- coherent_worker ( + modified_headers, + Nothing + ) + (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume + let + response_text = + serializeHTTPResponse response_headers fragments + + catch + (do + push_action response_text + return $ Just leftovers + ) + ((\ _e -> do + debugM "Session.HTTP1" "Session abandoned" + close_action + return Nothing + ) :: IOProblem -> IO (Maybe B.ByteString) ) + + HeadersAndBody_H1PC _headers _stopcondition _recv_leftovers -> do + -- print "HeadersAndBody_H1PC" + -- Let's see if I can go through the basic movements first, then through + -- more complicated things. + -- TODO: Implement posts and other requests with bodies.... + close_action + error "NotImplemented requests with bodies" + + +addExtraHeaders :: SessionsContext -> Headers -> Headers +addExtraHeaders sessions_context headers = + let + enriched_lens :: Lens' SessionsContext SessionsEnrichedHeaders + enriched_lens = (sessionsConfig . sessionsEnrichedHeaders) + -- Haskell laziness here! + headers_editor = He.fromList headers + -- TODO: Figure out which is the best way to put this contact in the + -- source code + protocol_lens = He.headerLens "second-transfer-eh--used-protocol" + add_used_protocol = sessions_context ^. (enriched_lens . addUsedProtocol ) + he1 = if add_used_protocol + then set protocol_lens (Just "HTTP/1.1") headers_editor + else headers_editor + result = He.toList he1 + + in if add_used_protocol + -- Nothing will be computed here if the headers are not modified. + then result + else headers diff --git a/hs-src/SecondTransfer/Http2/Framer.cpphs b/hs-src/SecondTransfer/Http2/Framer.cpphs deleted file mode 100644 --- a/hs-src/SecondTransfer/Http2/Framer.cpphs +++ /dev/null @@ -1,547 +0,0 @@ --- The framer has two functions: to convert bytes to Frames and the other way around, --- and two keep track of flow-control quotas. -{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, - DeriveDataTypeable, TemplateHaskell #-} -{-# OPTIONS_HADDOCK hide #-} -module SecondTransfer.Http2.Framer ( - BadPrefaceException, - - wrapSession, - http2FrameLength, - - -- Not needed anywhere, but supress the warning about unneeded symbol - closeAction - ) where - - - -import Control.Concurrent -import Control.Exception -import qualified Control.Exception as E -import Control.Lens (view) -import qualified Control.Lens as L -import Control.Monad.IO.Class (liftIO) -import qualified Control.Monad.Catch as C -import Control.Monad.Trans.Class (lift) -import Control.Monad.Trans.Reader -import Data.Binary (decode) -import qualified Data.ByteString as B -import qualified Data.ByteString.Lazy as LB -import Data.Conduit -import Data.Foldable (find) - -import qualified Network.HTTP2 as NH2 --- Logging utilities -import System.Log.Logger - -import qualified Data.HashTable.IO as H - -import SecondTransfer.Sessions.Internal (sessionExceptionHandler, nextSessionId, SessionsContext) -import SecondTransfer.Http2.Session -import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker) -import qualified SecondTransfer.MainLoop.Framer as F -import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction, - PullAction, PushAction) -import SecondTransfer.Utils (Word24, word24ToInt) -import SecondTransfer.Exception - - -#include "Logging.cpphs" - - -http2PrefixLength :: Int -http2PrefixLength = B.length NH2.connectionPreface - --- Let's do flow control here here .... - -type HashTable k v = H.CuckooHashTable k v - - -type GlobalStreamId = Int - - -data FlowControlCommand = - AddBytes_FCM Int - --- A hashtable from stream id to channel of availabiliy increases -type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand) - - -data CanOutput = CanOutput - - -data NoHeadersInChannel = NoHeadersInChannel - - -data FramerSessionData = FramerSessionData { - _stream2flow :: Stream2AvailSpace - , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString) - , _defaultStreamWindow :: MVar Int - - -- Wait variable to output bytes to the channel - , _canOutput :: MVar CanOutput - -- Flag that says if the session has been unwound... if such, - -- threads are adviced to exit as early as possible - , _outputIsForbidden :: MVar Bool - , _noHeadersInChannel :: MVar NoHeadersInChannel - , _pushAction :: PushAction - , _closeAction :: CloseAction - - -- Global id of the session, used for e.g. error reporting. - , _sessionId :: Int - - -- Sessions context, used for thing like e.g. error reporting - , _sessionsContext :: SessionsContext - - -- For GoAway frames - , _lastStream :: MVar Int - } - - -L.makeLenses ''FramerSessionData - - -type FramerSession = ReaderT FramerSessionData IO - - -wrapSession :: CoherentWorker -> SessionsContext -> Attendant -wrapSession coherent_worker sessions_context push_action pull_action close_action = do - - let - session_id_mvar = view nextSessionId sessions_context - - new_session_id <- modifyMVarMasked - session_id_mvar - (\ session_id -> return (session_id+1, session_id)) - - (session_input, session_output) <- (http2Session - coherent_worker new_session_id sessions_context) - - -- TODO : Add type annotations.... - s2f <- H.new - s2o <- H.new - default_stream_size_mvar <- newMVar 65536 - can_output <- newMVar CanOutput - no_headers_in_channel <- newMVar NoHeadersInChannel - last_stream_id <- newMVar 0 - output_is_forbidden <- newMVar False - - - -- We need some shared state - let framer_session_data = FramerSessionData { - _stream2flow = s2f - ,_stream2outputBytes = s2o - ,_defaultStreamWindow = default_stream_size_mvar - ,_canOutput = can_output - ,_noHeadersInChannel = no_headers_in_channel - ,_pushAction = push_action - ,_closeAction = close_action - ,_sessionId = new_session_id - ,_sessionsContext = sessions_context - ,_lastStream = last_stream_id - ,_outputIsForbidden = output_is_forbidden - } - - - let - -- TODO: Dodgy exception handling here... - close_on_error session_id session_context comp = - E.finally ( - E.catch - comp - (exc_handler session_id session_context) - ) - close_action - - exc_handler :: Int -> SessionsContext -> FramerException -> IO () - exc_handler x y e = do - modifyMVar_ output_is_forbidden (\ _ -> return True) - sessionExceptionHandler Framer_HTTP2SessionComponent x y e - - - forkIO - $ close_on_error new_session_id sessions_context - $ runReaderT (inputGatherer pull_action session_input ) framer_session_data - forkIO - $ close_on_error new_session_id sessions_context - $ runReaderT (outputGatherer session_output ) framer_session_data - - return () - - -http2FrameLength :: F.LengthCallback -http2FrameLength bs | (B.length bs) >= 3 = let - word24 = decode input_as_lbs :: Word24 - input_as_lbs = LB.fromStrict bs - in - Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses -http2FrameLength _ = Nothing - - -addCapacity :: - GlobalStreamId -> - Int -> - FramerSession () -addCapacity stream_id delta_cap = do - - if stream_id == 0 - then - -- TODO: Add session flow control - return () - else do - table <- view stream2flow - val <- liftIO $ H.lookup table stream_id - case val of - Nothing -> do - -- ?? - liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id) - (_, command_chan) <- startStreamOutputQueue stream_id - -- And try again - liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap - - - Just command_chan -> do - liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap - - --- This works by pulling bytes from the input side of the pipeline and converting them to frames. --- The frames are then put in the SessionInput. In the other end of the SessionInput they can be --- interpreted according to their HTTP/2 meaning. --- --- This function also does part of the flow control: it registers WindowUpdate frames and triggers --- quota updates on the streams. -inputGatherer :: PullAction -> SessionInput -> FramerSession () -inputGatherer pull_action session_input = do - -- We can start by reading off the prefix.... - INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Entering InputGatherer" ) - (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action - - if prefix /= NH2.connectionPreface - then do - sendGoAwayFrame NH2.ProtocolError - liftIO $ do - -- We just the the GoAway frame, although this is awfully early - -- and probably wrong - INSTRUMENTATION( errorM "HTTP2.Framer" "Invalid prologue") - throwIO BadPrefaceException - else - INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Prologue validated" ) - let - source::Source FramerSession B.ByteString - source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action - ( source $$ consume) - where - - consume :: Sink B.ByteString FramerSession () - consume = do - maybe_bytes <- await - -- Deserialize - - case maybe_bytes of - - Just bytes -> do - let - error_or_frame = NH2.decodeFrame some_settings bytes - -- TODO: See how we can change these.... - some_settings = NH2.defaultSettings - - case error_or_frame of - - Left _ -> do - -- Got an error from the decoder... meaning that a frame could - -- not be decoded.... in this case we send a cancel session command - -- to the session. - liftIO $ errorM "HTTP2.Framer" "CouldNotDecodeFrame" - -- Send frames like GoAway and such... - lift $ sendGoAwayFrame NH2.ProtocolError - -- Inform the session that it can tear down itself - liftIO $ sendCommandToSession session_input CancelSession_SIC - -- Any resources remaining here can be disposed - lift $ releaseFramer - -- And end this thread - - Right right_frame -> do - case right_frame of - - (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do - -- Bookkeep the increase on bytes on that stream - -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id) - lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit) - return () - - - frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do - -- Increase all the stuff.... - case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of - - Just (_, new_default_stream_size) -> do - old_default_stream_size_mvar <- view defaultStreamWindow - old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar - let general_delta = new_default_stream_size - old_default_stream_size - stream_to_flow <- view stream2flow - -- Add capacity to everybody's windows - liftIO $ - H.mapM_ ( - \ (k,v) -> if k /=0 - then writeChan v (AddBytes_FCM general_delta) - else return () ) - stream_to_flow - - - -- And set a new value - liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size - - - Nothing -> - -- This is a silenced internal error - return () - - -- And send the frame down to the session, so that session specific settings - -- can be applied. - liftIO $ sendFrameToSession session_input frame - - - a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do - -- Update the keep of last stream - lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id - - -- Send frame to the session - liftIO $ sendFrameToSession session_input a_frame - -- tail recursion: go again... - consume - - Nothing -> - -- We may as well exit this thread - return () - - --- All the output frames come this way first -outputGatherer :: SessionOutput -> FramerSession () -outputGatherer session_output = do - - -- We start by sending a settings frame - pushFrame - (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing) - (NH2.SettingsFrame []) - - loopPart - - where - - - dataForFrame p1 p2 = - LB.fromStrict $ NH2.encodeFrame p1 p2 - - loopPart :: FramerSession () - loopPart = do - - command_or_frame <- liftIO $ getFrameFromSession session_output - - case command_or_frame of - - Left CancelSession_SOC -> do - -- The session wants to cancel things - releaseFramer - - Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do - -- This frame is flow-controlled... I may be unable to send this frame in - -- some circumstances... - let stream_id = NH2.fromStreamIdentifier stream_idii - s2o <- view stream2outputBytes - lookup_result <- liftIO $ H.lookup s2o stream_id - stream_bytes_chan <- case lookup_result of - - Nothing -> do - (bc, _) <- startStreamOutputQueue stream_id - return bc - - Just bytes_chan -> return bytes_chan - - liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2 - - loopPart - - Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do - handleHeadersOfStream p1 p2 - - loopPart - - Right (p1, p2@(NH2.ContinuationFrame _) ) -> do - handleHeadersOfStream p1 p2 - - loopPart - - Right (p1, p2) -> do - -- Most other frames go right away... as long as no headers are in process... - no_headers <- view noHeadersInChannel - liftIO $ takeMVar no_headers - pushFrame p1 p2 - liftIO $ putMVar no_headers NoHeadersInChannel - - loopPart - - -updateLastStream :: GlobalStreamId -> FramerSession () -updateLastStream stream_id = do - last_stream_id_mvar <- view lastStream - liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) - - - -startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand) -startStreamOutputQueue stream_id = do - -- New thread for handling outputs of this stream is needed - bytes_chan <- liftIO newChan - command_chan <- liftIO newChan - - s2o <- view stream2outputBytes - - liftIO $ H.insert s2o stream_id bytes_chan - - s2c <- view stream2flow - - - liftIO $ H.insert s2c stream_id command_chan - - -- - initial_cap_mvar <- view defaultStreamWindow - initial_cap <- liftIO $ readMVar initial_cap_mvar - close_action <- view closeAction - sessions_context <- view sessionsContext - session_id' <- view SecondTransfer.Http2.Framer.sessionId - output_is_forbidden_mvar <- view outputIsForbidden - - -- And don't forget the thread itself - let - close_on_error session_id session_context comp = - E.finally - (E.catch - comp - (exc_handler session_id session_context) - ) - close_action - - exc_handler :: Int -> SessionsContext -> IOProblem -> IO () - exc_handler x y e = do - -- Let's also decree that other streams don't even try - modifyMVar_ output_is_forbidden_mvar ( \ _ -> return True) - sessionExceptionHandler Framer_HTTP2SessionComponent x y e - - - read_state <- ask - liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT - (flowControlOutput stream_id initial_cap "" command_chan bytes_chan) - read_state - - return (bytes_chan , command_chan) - - --- This works in the output side of the HTTP/2 framing session, and it acts as a --- semaphore ensuring that headers are output without any interleaved frames. --- --- There are more synchronization mechanisms in the session, this does not act --- alone. -handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession () -handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload - | (frameIsHeaderOfStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do - -- Take it - no_headers <- view noHeadersInChannel - liftIO $ takeMVar no_headers - pushFrame p1 frame_payload - -- DONT PUT THE MvAR HERE - - | (frameIsHeaderOfStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do - no_headers <- view noHeadersInChannel - liftIO $ takeMVar no_headers - pushFrame p1 frame_payload - -- Since we finish.... - liftIO $ putMVar no_headers NoHeadersInChannel - - | frameEndsHeaders p1 frame_payload = do - -- I can only get here for a continuation frame after something else that is a headers - no_headers <- view noHeadersInChannel - liftIO $ putMVar no_headers NoHeadersInChannel - return () - - -frameIsHeaderOfStream :: NH2.FramePayload -> Bool -frameIsHeaderOfStream (NH2.HeadersFrame _ _ ) - = True -frameIsHeaderOfStream _ - = False - - -frameEndsHeaders :: NH2.EncodeInfo -> NH2.FramePayload -> Bool -frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags -frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags -frameEndsHeaders _ _ = False - - --- Push a frame into the output channel... this waits for the --- channel to be free to send. -pushFrame :: NH2.EncodeInfo - -> NH2.FramePayload -> FramerSession () -pushFrame p1 p2 = do - let bs = LB.fromStrict $ NH2.encodeFrame p1 p2 - sendBytes bs - - -sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession () -sendGoAwayFrame error_code = do - last_stream_id_mvar <- view lastStream - last_stream_id <- liftIO $ readMVar last_stream_id_mvar - pushFrame (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing) - (NH2.GoAwayFrame (NH2.toStreamIdentifier last_stream_id) error_code "") - - -sendBytes :: LB.ByteString -> FramerSession () -sendBytes bs = do - push_action <- view pushAction - can_output <- view canOutput - liftIO $ do - bs `seq` - (C.bracket - (takeMVar can_output) - (\ _ -> push_action bs) - (\ c -> putMVar can_output c) - ) - - --- A thread in charge of doing flow control transmission....This sends already --- formatted frames (ByteStrings), not the frames themselves. And it doesn't --- mess with the structure of the packets. -flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession () -flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do - if leftovers == "" - then do - -- Get more data (possibly block waiting for it) - bytes_to_send <- liftIO $ readChan bytes_chan - flowControlOutput stream_id capacity bytes_to_send commands_chan bytes_chan - else do - -- Length? - let amount = fromIntegral $ ((LB.length leftovers) - 9) - if amount <= capacity - then do - -- Is - -- I can send ... if no headers are in process.... - no_headers <- view noHeadersInChannel - C.bracket - (liftIO $ takeMVar no_headers) - (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel) - (\ _ -> sendBytes leftovers ) - flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan - else do - -- I can not send because flow-control is full, wait for a command instead - -- liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id) - command <- liftIO $ readChan commands_chan - case command of - AddBytes_FCM delta_cap -> do - -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id) - flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan - - -releaseFramer :: FramerSession () -releaseFramer = do - -- Release any resources pending... - - return () diff --git a/hs-src/SecondTransfer/Http2/Framer.hs b/hs-src/SecondTransfer/Http2/Framer.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Framer.hs @@ -0,0 +1,562 @@ +-- The framer has two functions: to convert bytes to Frames and the other way around, +-- and two keep track of flow-control quotas. +{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, + DeriveDataTypeable, TemplateHaskell #-} +{-# OPTIONS_HADDOCK hide #-} +module SecondTransfer.Http2.Framer ( + BadPrefaceException, + + wrapSession, + http2FrameLength, + + -- Not needed anywhere, but supress the warning about unneeded symbol + closeAction + ) where + + + +import Control.Concurrent +import Control.Exception +import qualified Control.Exception as E +import Control.Lens (view) +import qualified Control.Lens as L +import Control.Monad.IO.Class (liftIO) +import qualified Control.Monad.Catch as C +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Reader +import Data.Binary (decode) +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as LB +import Data.Conduit +import Data.Foldable (find) + +import qualified Network.HTTP2 as NH2 +-- Logging utilities +import System.Log.Logger + +import qualified Data.HashTable.IO as H + +import SecondTransfer.Sessions.Internal (sessionExceptionHandler, nextSessionId, SessionsContext) +import SecondTransfer.Http2.Session +import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker) +import qualified SecondTransfer.MainLoop.Framer as F +import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction, + PullAction, PushAction) +import SecondTransfer.Utils (Word24, word24ToInt) +import SecondTransfer.Exception + + +#include "Logging.cpphs" + + +http2PrefixLength :: Int +http2PrefixLength = B.length NH2.connectionPreface + +-- Let's do flow control here here .... + +type HashTable k v = H.CuckooHashTable k v + + +type GlobalStreamId = Int + + +data FlowControlCommand = + AddBytes_FCM Int + +-- A hashtable from stream id to channel of availabiliy increases +type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand) + + +data CanOutput = CanOutput + + +data NoHeadersInChannel = NoHeadersInChannel + + +data FramerSessionData = FramerSessionData { + _stream2flow :: Stream2AvailSpace + , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString) + , _defaultStreamWindow :: MVar Int + + -- Wait variable to output bytes to the channel + , _canOutput :: MVar CanOutput + -- Flag that says if the session has been unwound... if such, + -- threads are adviced to exit as early as possible + , _outputIsForbidden :: MVar Bool + , _noHeadersInChannel :: MVar NoHeadersInChannel + , _pushAction :: PushAction + , _closeAction :: CloseAction + + -- Global id of the session, used for e.g. error reporting. + , _sessionId :: Int + + -- Sessions context, used for thing like e.g. error reporting + , _sessionsContext :: SessionsContext + + -- For GoAway frames + , _lastStream :: MVar Int + } + + +L.makeLenses ''FramerSessionData + + +type FramerSession = ReaderT FramerSessionData IO + + +wrapSession :: CoherentWorker -> SessionsContext -> Attendant +wrapSession coherent_worker sessions_context push_action pull_action close_action = do + + let + session_id_mvar = view nextSessionId sessions_context + + new_session_id <- modifyMVarMasked + session_id_mvar + (\ session_id -> return (session_id+1, session_id)) + + (session_input, session_output) <- (http2Session + coherent_worker new_session_id sessions_context) + + -- TODO : Add type annotations.... + s2f <- H.new + s2o <- H.new + default_stream_size_mvar <- newMVar 65536 + can_output <- newMVar CanOutput + no_headers_in_channel <- newMVar NoHeadersInChannel + last_stream_id <- newMVar 0 + output_is_forbidden <- newMVar False + + + -- We need some shared state + let framer_session_data = FramerSessionData { + _stream2flow = s2f + ,_stream2outputBytes = s2o + ,_defaultStreamWindow = default_stream_size_mvar + ,_canOutput = can_output + ,_noHeadersInChannel = no_headers_in_channel + ,_pushAction = push_action + ,_closeAction = close_action + ,_sessionId = new_session_id + ,_sessionsContext = sessions_context + ,_lastStream = last_stream_id + ,_outputIsForbidden = output_is_forbidden + } + + + let + -- TODO: Dodgy exception handling here... + close_on_error session_id session_context comp = + E.finally ( + E.catch( + ( + E.catch + comp + (exc_handler session_id session_context) + ) + ) + (io_exc_handler session_id session_context) + ) + close_action + + exc_handler :: Int -> SessionsContext -> FramerException -> IO () + exc_handler x y e = do + modifyMVar_ output_is_forbidden (\ _ -> return True) + errorM "HTTP2.Framer" "Exception went up" + sessionExceptionHandler Framer_HTTP2SessionComponent x y e + + io_exc_handler :: Int -> SessionsContext -> IOProblem -> IO () + io_exc_handler x y e = do + modifyMVar_ output_is_forbidden (\ _ -> return True) + -- !!! These exceptions are way too common for we to care.... + -- errorM "HTTP2.Framer" "Exception went up" + -- sessionExceptionHandler Framer_HTTP2SessionComponent x y e + + + forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (inputGatherer pull_action session_input ) framer_session_data + forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (outputGatherer session_output ) framer_session_data + + return () + + +http2FrameLength :: F.LengthCallback +http2FrameLength bs | (B.length bs) >= 3 = let + word24 = decode input_as_lbs :: Word24 + input_as_lbs = LB.fromStrict bs + in + Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses +http2FrameLength _ = Nothing + + +addCapacity :: + GlobalStreamId -> + Int -> + FramerSession () +addCapacity stream_id delta_cap = do + + if stream_id == 0 + then + -- TODO: Add session flow control + return () + else do + table <- view stream2flow + val <- liftIO $ H.lookup table stream_id + case val of + Nothing -> do + -- TODO: There is a very important compliance note here. Fix this, otherwise + -- a peer can create a memory overflow by sending WindowUpdate on unexistent + -- streams. + liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id) + (_, command_chan) <- startStreamOutputQueue stream_id + -- And try again + liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap + + + Just command_chan -> do + liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap + + +-- This works by pulling bytes from the input side of the pipeline and converting them to frames. +-- The frames are then put in the SessionInput. In the other end of the SessionInput they can be +-- interpreted according to their HTTP/2 meaning. +-- +-- This function also does part of the flow control: it registers WindowUpdate frames and triggers +-- quota updates on the streams. +inputGatherer :: PullAction -> SessionInput -> FramerSession () +inputGatherer pull_action session_input = do + -- We can start by reading off the prefix.... + (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action + + if prefix /= NH2.connectionPreface + then do + sendGoAwayFrame NH2.ProtocolError + liftIO $ do + -- We just the the GoAway frame, although this is awfully early + -- and probably wrong + INSTRUMENTATION( errorM "HTTP2.Framer" "Invalid prologue") + throwIO BadPrefaceException + else + INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Prologue validated" ) + let + source::Source FramerSession B.ByteString + source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action + ( source $$ consume) + where + + consume :: Sink B.ByteString FramerSession () + consume = do + maybe_bytes <- await + -- Deserialize + + case maybe_bytes of + + Just bytes -> do + let + error_or_frame = NH2.decodeFrame some_settings bytes + -- TODO: See how we can change these.... + some_settings = NH2.defaultSettings + + case error_or_frame of + + Left _ -> do + -- Got an error from the decoder... meaning that a frame could + -- not be decoded.... in this case we send a cancel session command + -- to the session. + liftIO $ errorM "HTTP2.Framer" "CouldNotDecodeFrame" + -- Send frames like GoAway and such... + lift $ sendGoAwayFrame NH2.ProtocolError + -- Inform the session that it can tear down itself + liftIO $ sendCommandToSession session_input CancelSession_SIC + -- Any resources remaining here can be disposed + lift $ releaseFramer + -- And end this thread + + Right right_frame -> do + case right_frame of + + (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do + -- Bookkeep the increase on bytes on that stream + -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id) + lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit) + return () + + + frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do + -- Increase all the stuff.... + case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of + + Just (_, new_default_stream_size) -> do + old_default_stream_size_mvar <- view defaultStreamWindow + old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar + let general_delta = new_default_stream_size - old_default_stream_size + stream_to_flow <- view stream2flow + -- Add capacity to everybody's windows + liftIO $ + H.mapM_ ( + \ (k,v) -> if k /=0 + then writeChan v (AddBytes_FCM general_delta) + else return () ) + stream_to_flow + + + -- And set a new value + liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size + + + Nothing -> + -- This is a silenced internal error + return () + + -- And send the frame down to the session, so that session specific settings + -- can be applied. + liftIO $ sendFrameToSession session_input frame + + + a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do + -- Update the keep of last stream + lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id + + -- Send frame to the session + liftIO $ sendFrameToSession session_input a_frame + -- tail recursion: go again... + consume + + Nothing -> + -- We may as well exit this thread + return () + + +-- All the output frames come this way first +outputGatherer :: SessionOutput -> FramerSession () +outputGatherer session_output = do + + -- We start by sending a settings frame + pushFrame + (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing) + (NH2.SettingsFrame []) + + loopPart + + where + + + dataForFrame p1 p2 = + LB.fromStrict $ NH2.encodeFrame p1 p2 + + loopPart :: FramerSession () + loopPart = do + + command_or_frame <- liftIO $ getFrameFromSession session_output + + case command_or_frame of + + Left CancelSession_SOC -> do + -- The session wants to cancel things + INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "CancelSession_SOC processed") + releaseFramer + + Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do + -- This frame is flow-controlled... I may be unable to send this frame in + -- some circumstances... + let stream_id = NH2.fromStreamIdentifier stream_idii + s2o <- view stream2outputBytes + lookup_result <- liftIO $ H.lookup s2o stream_id + stream_bytes_chan <- case lookup_result of + + Nothing -> do + (bc, _) <- startStreamOutputQueue stream_id + return bc + + Just bytes_chan -> return bytes_chan + + liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2 + loopPart + + Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do + handleHeadersOfStream p1 p2 + loopPart + + Right (p1, p2@(NH2.ContinuationFrame _) ) -> do + handleHeadersOfStream p1 p2 + loopPart + + Right (p1, p2) -> do + -- Most other frames go right away... as long as no headers are in process... + no_headers <- view noHeadersInChannel + liftIO $ takeMVar no_headers + pushFrame p1 p2 + liftIO $ putMVar no_headers NoHeadersInChannel + loopPart + + +updateLastStream :: GlobalStreamId -> FramerSession () +updateLastStream stream_id = do + last_stream_id_mvar <- view lastStream + liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) + + + +startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand) +startStreamOutputQueue stream_id = do + -- New thread for handling outputs of this stream is needed + bytes_chan <- liftIO newChan + command_chan <- liftIO newChan + + s2o <- view stream2outputBytes + + liftIO $ H.insert s2o stream_id bytes_chan + + s2c <- view stream2flow + + + liftIO $ H.insert s2c stream_id command_chan + + -- + initial_cap_mvar <- view defaultStreamWindow + initial_cap <- liftIO $ readMVar initial_cap_mvar + close_action <- view closeAction + sessions_context <- view sessionsContext + session_id' <- view SecondTransfer.Http2.Framer.sessionId + output_is_forbidden_mvar <- view outputIsForbidden + + -- And don't forget the thread itself + let + close_on_error session_id session_context comp = + E.finally + (E.catch + comp + (exc_handler session_id session_context) + ) + close_action + + exc_handler :: Int -> SessionsContext -> IOProblem -> IO () + exc_handler x y e = do + -- Let's also decree that other streams don't even try + modifyMVar_ output_is_forbidden_mvar ( \ _ -> return True) + sessionExceptionHandler Framer_HTTP2SessionComponent x y e + + + read_state <- ask + liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT + (flowControlOutput stream_id initial_cap "" command_chan bytes_chan) + read_state + + return (bytes_chan , command_chan) + + +-- This works in the output side of the HTTP/2 framing session, and it acts as a +-- semaphore ensuring that headers are output without any interleaved frames. +-- +-- There are more synchronization mechanisms in the session... +handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession () +handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload + | (frameIsHeadersAndOpensStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do + -- Take it + no_headers <- view noHeadersInChannel + liftIO $ takeMVar no_headers + pushFrame p1 frame_payload + -- DONT PUT THE MvAR HERE + + | (frameIsHeadersAndOpensStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do + no_headers <- view noHeadersInChannel + liftIO $ takeMVar no_headers + pushFrame p1 frame_payload + -- Since we finish.... + liftIO $ putMVar no_headers NoHeadersInChannel + + | frameEndsHeaders p1 frame_payload = do + -- I can only get here for a continuation frame after something else that is a headers + no_headers <- view noHeadersInChannel + pushFrame p1 frame_payload + liftIO $ putMVar no_headers NoHeadersInChannel + return () + + | otherwise = do + -- Nothing to do with the mvar, the no_headers should be empty + pushFrame p1 frame_payload + + +frameIsHeadersAndOpensStream :: NH2.FramePayload -> Bool +frameIsHeadersAndOpensStream (NH2.HeadersFrame _ _ ) + = True +frameIsHeadersAndOpensStream _ + = False + + +frameEndsHeaders :: NH2.EncodeInfo -> NH2.FramePayload -> Bool +frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags +frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags +frameEndsHeaders _ _ = False + + +-- Push a frame into the output channel... this waits for the +-- channel to be free to send. +pushFrame :: NH2.EncodeInfo + -> NH2.FramePayload -> FramerSession () +pushFrame p1 p2 = do + let bs = LB.fromStrict $ NH2.encodeFrame p1 p2 + sendBytes bs + + +sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession () +sendGoAwayFrame error_code = do + last_stream_id_mvar <- view lastStream + last_stream_id <- liftIO $ readMVar last_stream_id_mvar + pushFrame (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing) + (NH2.GoAwayFrame (NH2.toStreamIdentifier last_stream_id) error_code "") + + +sendBytes :: LB.ByteString -> FramerSession () +sendBytes bs = do + push_action <- view pushAction + can_output <- view canOutput + liftIO $ do + bs `seq` + (C.bracket + (takeMVar can_output) + (\ _ -> push_action bs) + (\ c -> putMVar can_output c) + ) + + +-- A thread in charge of doing flow control transmission....This sends already +-- formatted frames (ByteStrings), not the frames themselves. And it doesn't +-- mess with the structure of the packets. +flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession () +flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do + if leftovers == "" + then do + -- Get more data (possibly block waiting for it) + bytes_to_send <- liftIO $ readChan bytes_chan + flowControlOutput stream_id capacity bytes_to_send commands_chan bytes_chan + else do + -- Length? + let amount = fromIntegral $ ((LB.length leftovers) - 9) + if amount <= capacity + then do + -- Is + -- I can send ... if no headers are in process.... + no_headers <- view noHeadersInChannel + C.bracket + (liftIO $ takeMVar no_headers) + (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel) + (\ _ -> sendBytes leftovers ) + flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan + else do + -- I can not send because flow-control is full, wait for a command instead + -- liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id) + command <- liftIO $ readChan commands_chan + case command of + AddBytes_FCM delta_cap -> do + -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id) + flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan + + +releaseFramer :: FramerSession () +releaseFramer = do + -- Release any resources pending... + + return () diff --git a/hs-src/SecondTransfer/Http2/Session.cpphs b/hs-src/SecondTransfer/Http2/Session.cpphs deleted file mode 100644 --- a/hs-src/SecondTransfer/Http2/Session.cpphs +++ /dev/null @@ -1,754 +0,0 @@ --- Session: links frames to streams, and helps in ordering the header frames --- so that they don't get mixed with header frames from other streams when --- resources are being served concurrently. -{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-} -{-# OPTIONS_HADDOCK hide #-} -module SecondTransfer.Http2.Session( - http2Session - ,getFrameFromSession - ,sendFrameToSession - ,sendCommandToSession - - ,CoherentSession - ,SessionInput(..) - ,SessionInputCommand(..) - ,SessionOutput(..) - ,SessionOutputCommand(..) - ,SessionCoordinates(..) - ,SessionComponent(..) - ,SessionsCallbacks(..) - ,SessionsConfig(..) - ,ErrorCallback - - -- Internal stuff - ,OutputFrame - ,InputFrame - ) where - -#include "Logging.cpphs" - --- System grade utilities -import Control.Concurrent (ThreadId, forkIO) -import Control.Concurrent.Chan -import Control.Exception (throwTo) -import qualified Control.Exception as E -import Control.Monad (forever) -import Control.Monad.IO.Class (liftIO) -import Control.Monad.Trans.Reader - -import Control.Concurrent.MVar -import qualified Data.ByteString as B -import qualified Data.ByteString.Builder as Bu -import qualified Data.ByteString.Lazy as Bl -import Data.Conduit -import qualified Data.HashTable.IO as H -import qualified Data.IntSet as NS -import Data.Monoid as Mo - -import Control.Lens - --- No framing layer here... let's use Kazu's Yamamoto library -import qualified Network.HPACK as HP -import qualified Network.HTTP2 as NH2 - --- Logging utilities -import System.Log.Logger - --- Imports from other parts of the program -import SecondTransfer.MainLoop.CoherentWorker -import SecondTransfer.MainLoop.Tokens -import SecondTransfer.Sessions.Config -import SecondTransfer.Sessions.Internal (sessionExceptionHandler, SessionsContext) -import SecondTransfer.Utils (unfoldChannelAndSource) -import SecondTransfer.Exception - --- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to --- use :-( -type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload) -type InputFrame = NH2.Frame - - -useChunkLength :: Int -useChunkLength = 16384 - - --- Singleton instance used for concurrency -data HeadersSent = HeadersSent - --- All streams put their data bits here. A "Nothing" value signals --- end of data. -type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString) - - --- Whatever a worker thread is going to need comes here.... --- this is to make refactoring easier, but not strictly needed. -data WorkerThreadEnvironment = WorkerThreadEnvironment { - -- What's the header stream id? - _streamId :: GlobalStreamId - - -- A full block of headers can come here... the mvar in the middle should - -- be populate to signal end of headers transmission. A thread will be suspended - -- waiting for that - , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers) - - -- And regular contents can come this way and thus be properly mixed - -- with everything else.... for now... - ,_dataOutput :: Chan DataOutputToConveyor - - ,_streamsCancelled_WTE :: MVar NS.IntSet - - } - -makeLenses ''WorkerThreadEnvironment - - --- An HTTP/2 session. Basically a couple of channels ... -type Session = (SessionInput, SessionOutput) - - --- From outside, one can only write to this one ... the newtype is to enforce --- this. -newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) ) -sendFrameToSession :: SessionInput -> InputFrame -> IO () -sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame - -sendCommandToSession :: SessionInput -> SessionInputCommand -> IO () -sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command - --- From outside, one can only read from this one -newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) ) -getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) -getFrameFromSession (SessionOutput chan) = readChan chan - - -type HashTable k v = H.CuckooHashTable k v - - --- Blaze builder could be more proper here... -type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder - - -type WorkerMonad = ReaderT WorkerThreadEnvironment IO - - --- Have to figure out which are these...but I would expect to have things --- like unexpected aborts here in this type. -data SessionInputCommand = - CancelSession_SIC - deriving Show - - --- temporary -data SessionOutputCommand = - CancelSession_SOC - deriving Show - - --- Here is how we make a session -type SessionMaker = SessionsContext -> IO Session - - --- Here is how we make a session wrapping a CoherentWorker -type CoherentSession = CoherentWorker -> SessionMaker - - -data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream) - - --- NH2.Frame != Frame -data SessionData = SessionData { - -- ATTENTION: Ignore the warning coming from here for now - _sessionsContext :: SessionsContext - - ,_sessionInput :: Chan (Either SessionInputCommand InputFrame) - - -- We need to lock this channel occassionally so that we can order multiple - -- header frames properly.... - ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame)) - - -- Use to encode - ,_toEncodeHeaders :: MVar HP.DynamicTable - -- And used to decode - ,_toDecodeHeaders :: MVar HP.DynamicTable - - -- Used for decoding the headers - ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment - - -- Used for worker threads... this is actually a pre-filled template - -- I make copies of it in different contexts, and as needed. - ,_forWorkerThread :: WorkerThreadEnvironment - - ,_coherentWorker :: CoherentWorker - - -- Some streams may be cancelled - ,_streamsCancelled :: MVar NS.IntSet - - -- Data input mechanism corresponding to some threads - ,_stream2PostInputMechanism :: HashTable Int PostInputMechanism - - -- Worker thread register. This is a dictionary from stream id to - -- the ThreadId of the thread with the worker thread. I use this to - -- raise asynchronous exceptions in the worker thread if the stream - -- is cancelled by the client. This way we get early finalization. - ,_stream2WorkerThread :: HashTable Int ThreadId - - -- Use to retrieve/set the session id - ,_sessionIdAtSession :: Int - } - - -makeLenses ''SessionData - - --- v- {headers table size comes here!!} -http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session -http2Session coherent_worker session_id sessions_context = do - session_input <- newChan - session_output <- newChan - session_output_mvar <- newMVar session_output - - - -- For incremental construction of headers... - stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment - - -- Warning: we should find a way of coping with different table sizes. - decode_headers_table <- HP.newDynamicTableForDecoding 4096 - decode_headers_table_mvar <- newMVar decode_headers_table - - encode_headers_table <- HP.newDynamicTableForEncoding 4096 - encode_headers_table_mvar <- newMVar encode_headers_table - - -- These ones need independent threads taking care of sending stuff - -- their way... - headers_output <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers)) - data_output <- newChan :: IO (Chan DataOutputToConveyor) - - stream2postinputmechanism <- H.new - stream2workerthread <- H.new - - -- What about stream cancellation? - cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet) - - let for_worker_thread = WorkerThreadEnvironment { - _streamId = error "NotInitialized" - ,_headersOutput = headers_output - ,_dataOutput = data_output - ,_streamsCancelled_WTE = cancelled_streams_mvar - } - - let session_data = SessionData { - _sessionsContext = sessions_context - ,_sessionInput = session_input - ,_sessionOutput = session_output_mvar - ,_toDecodeHeaders = decode_headers_table_mvar - ,_toEncodeHeaders = encode_headers_table_mvar - ,_stream2HeaderBlockFragment = stream_request_headers - ,_forWorkerThread = for_worker_thread - ,_coherentWorker = coherent_worker - ,_streamsCancelled = cancelled_streams_mvar - ,_stream2PostInputMechanism = stream2postinputmechanism - ,_stream2WorkerThread = stream2workerthread - ,_sessionIdAtSession = session_id - } - - let - exc_handler :: SessionComponent -> HTTP2SessionException -> IO () - exc_handler component e = sessionExceptionHandler component session_id sessions_context e - exc_guard :: SessionComponent -> IO () -> IO () - exc_guard component action = E.catch action $ exc_handler component - - -- Create an input thread that decodes frames... - forkIO $ exc_guard SessionInputThread_HTTP2SessionComponent - $ runReaderT sessionInputThread session_data - - -- Create a thread that captures headers and sends them down the tube - forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent - $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data - - -- Create a thread that captures data and sends it down the tube - forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent - $ dataOutputThread data_output session_output_mvar - - -- The two previous threads fill the session_output argument below (they write to it) - -- the session machinery in the other end is in charge of sending that data through the - -- socket. - - return ( (SessionInput session_input), - (SessionOutput session_output) ) - - --- TODO: Some ill clients can break this thread with exceptions. Make these paths a bit ---- more robust. -sessionInputThread :: ReaderT SessionData IO () -sessionInputThread = do - INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread" ) - - -- This is an introductory and declarative block... all of this is tail-executed - -- every time that a packet needs to be processed. It may be a good idea to abstract - -- these values in a closure... - session_input <- view sessionInput - - decode_headers_table_mvar <- view toDecodeHeaders - stream_request_headers <- view stream2HeaderBlockFragment - cancelled_streams_mvar <- view streamsCancelled - coherent_worker <- view coherentWorker - - for_worker_thread_uns <- view forWorkerThread - stream2workerthread <- view stream2WorkerThread - - input <- liftIO $ readChan session_input - - INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input) ) - - case input of - - Left CancelSession_SIC -> do - -- Good place to tear down worker threads... Let the rest of the finalization - -- to the framer - liftIO $ do - H.mapM_ - (\ (_, thread_id) -> do - throwTo thread_id StreamCancelledException - infoM "HTTP2.Session" $ "Stream successfully interrupted" - ) - stream2workerthread - - -- We do not continue here, but instead let it finish - return () - - Right frame | Just (stream_id, bytes) <- frameIsHeaderOfStream frame -> do - -- Just append the frames to streamRequestHeaders - appendHeaderFragmentBlock stream_id bytes - - if frameEndsHeaders frame then - do - -- Let's decode the headers - let for_worker_thread = set streamId stream_id for_worker_thread_uns - headers_bytes <- getHeaderBytes stream_id - dyn_table <- liftIO $ takeMVar decode_headers_table_mvar - (new_table, header_list ) <- liftIO $ {-# SCC "decodeHeader" #-} HP.decodeHeader dyn_table headers_bytes - -- Good moment to remove the headers from the table.... we don't want a space - -- leak here - liftIO $ do - H.delete stream_request_headers stream_id - putMVar decode_headers_table_mvar new_table - - -- TODO: Validate headers, abort session if the headers are invalid. - -- Otherwise other invariants will break!! - -- THIS IS PROBABLY THE BEST PLACE FOR DOING IT. - - -- If the headers end the request.... - post_data_source <- if not (frameEndsStream frame) - then do - mechanism <- createMechanismForStream stream_id - let source = postDataSourceFromMechanism mechanism - return $ Just source - else do - return Nothing - - -- I'm clear to start the worker, in its own thread - liftIO $ do - thread_id <- forkIO $ runReaderT - (workerThread (header_list, post_data_source) coherent_worker) - for_worker_thread - H.insert stream2workerthread stream_id thread_id - - return () - else - -- Frame doesn't end the headers... it was added before... so - -- probably do nothing - return () - - continue - - Right frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do - let stream_id = streamIdFromFrame frame - liftIO $ do - INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show _error_code_id) ) - cancelled_streams <- takeMVar cancelled_streams_mvar - INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) ) - putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams - maybe_thread_id <- H.lookup stream2workerthread stream_id - case maybe_thread_id of - Nothing -> - -- This is actually more like an internal error - error "InterruptingUnexistentStream" - - Just thread_id -> do - throwTo thread_id StreamCancelledException - INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" ) - - continue - - Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) -> do - -- So I got data to process - -- TODO: Handle end of stream - let stream_id = NH2.fromStreamIdentifier nh2_stream_id - -- TODO: Handle the cases where the stream_id doesn't match an already existent - -- stream. In such cases it is justified to reset the connection with a protocol_error. - - streamWorkerSendData stream_id somebytes - -- After that data has been received and forwarded downstream, we can issue a windows update - -- - -- TODO: We can use wider frames to avoid congestion... - -- .... and we can also be more compositional with these short bursts of data.... - -- - -- TODO: Consider that the best place to output these frames can be somewhere else... - -- - -- TODO: Use a special, with-quota queue here to do flow control. Don't send meaningless - -- WindowUpdateFrame's - sendOutFrame - (NH2.EncodeInfo - NH2.defaultFlags - nh2_stream_id - Nothing - ) - (NH2.WindowUpdateFrame - (fromIntegral (B.length somebytes)) - ) - sendOutFrame - (NH2.EncodeInfo - NH2.defaultFlags - (NH2.toStreamIdentifier 0) - Nothing - ) - (NH2.WindowUpdateFrame - (fromIntegral (B.length somebytes)) - ) - - if frameEndsStream frame - then do - -- Good place to close the source ... - closePostDataSource stream_id - else - return () - - continue - - Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do - -- Deal with pings: this is an Ack, so do nothing - continue - - Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do - -- Deal with pings: NOT an Ack, so answer - INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Ping processed" ) - sendOutFrame - (NH2.EncodeInfo - (NH2.setAck NH2.defaultFlags) - (NH2.toStreamIdentifier 0) - Nothing - ) - (NH2.PingFrame somebytes) - - continue - - Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do - -- Frame was received by the peer, do nothing here... - continue - - -- TODO: Do something with these settings!! - Right (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do - INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) ) - -- Just acknowledge the frame.... for now - sendOutFrame - (NH2.EncodeInfo - (NH2.setAck NH2.defaultFlags) - (NH2.toStreamIdentifier 0) - Nothing ) - (NH2.SettingsFrame []) - - continue - - - Right somethingelse -> do - -- An undhandled case here.... - INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: " ) - liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse) - - continue - - where - - continue = sessionInputThread - - sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO () - sendOutFrame encode_info payload = do - session_output_mvar <- view sessionOutput - - session_output <- liftIO $ takeMVar session_output_mvar - liftIO $ writeChan session_output $ Right (encode_info, payload) - liftIO $ putMVar session_output_mvar session_output - - -frameEndsStream :: InputFrame -> Bool -frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags - - -createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism -createMechanismForStream stream_id = do - (chan, source) <- liftIO $ unfoldChannelAndSource - stream2postinputmechanism <- view stream2PostInputMechanism - let pim = PostInputMechanism (chan, source) - liftIO $ H.insert stream2postinputmechanism stream_id pim - return pim - - --- TODO: Can be optimized by factoring out the mechanism lookup --- TODO IMPORTANT: This is a good place to drop the postinputmechanism --- for a stream, so that unprocessed data can be garbage-collected. -closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO () -closePostDataSource stream_id = do - stream2postinputmechanism <- view stream2PostInputMechanism - - pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id - - case pim_maybe of - - Just (PostInputMechanism (chan, _)) -> - liftIO $ writeChan chan Nothing - - Nothing -> - -- TODO: This is a protocol error, handle it properly - error "Internal error/closePostDataSource" - - -streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO () -streamWorkerSendData stream_id bytes = do - s2pim <- view stream2PostInputMechanism - pim_maybe <- liftIO $ H.lookup s2pim stream_id - - case pim_maybe of - - Just pim -> - sendBytesToPim pim bytes - - Nothing -> - -- This is an internal error, the mechanism should be - -- created when the headers end (and if the headers - -- do not finish the stream) - error "Internal error" - - -sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO () -sendBytesToPim (PostInputMechanism (chan, _)) bytes = - liftIO $ writeChan chan (Just bytes) - - -postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream -postDataSourceFromMechanism (PostInputMechanism (_, source)) = source - - -isSettingsAck :: NH2.FrameHeader -> Bool -isSettingsAck (NH2.FrameHeader _ flags _) = - NH2.testAck flags - - -isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool -isStreamCancelled stream_id = do - cancelled_streams_mvar <- view streamsCancelled_WTE - cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar - return $ NS.member stream_id cancelled_streams - - - -workerThread :: Request -> CoherentWorker -> WorkerMonad () -workerThread req coherent_worker = - do - headers_output <- view headersOutput - stream_id <- view streamId - - (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req - - -- Now I send the headers, if that's possible at all - headers_sent <- liftIO $ newEmptyMVar - liftIO $ writeChan headers_output (stream_id, headers_sent, headers) - - -- At this moment I should ask if the stream hasn't been cancelled by the browser before - -- commiting to the work of sending addtitional data - is_stream_cancelled <- isStreamCancelled stream_id - if not is_stream_cancelled - - then do - -- I have a beautiful source that I can de-construct... - -- TODO: Optionally pulling data out from a Conduit .... - -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) ) - -- - -- This threadlet should block here waiting for the headers to finish going - (maybe_footers, _) <- runConduit $ - (transPipe liftIO data_and_conclussion) - `fuseBothMaybe` - (sendDataOfStream stream_id headers_sent) - -- BIG TODO: Send the footers ... likely stream conclusion semantics - -- will need to be changed. - return () - else - - return () - --- v-- comp. monad. -sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) () -sendDataOfStream stream_id headers_sent = do - data_output <- view dataOutput - -- Wait for all headers sent - liftIO $ takeMVar headers_sent - consumer data_output - where - consumer data_output = do - maybe_bytes <- await - case maybe_bytes of - Nothing -> - liftIO $ writeChan data_output (stream_id, Nothing) - Just bytes -> do - liftIO $ writeChan data_output (stream_id, Just bytes) - consumer data_output - - -appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO () -appendHeaderFragmentBlock global_stream_id bytes = do - ht <- view stream2HeaderBlockFragment - maybe_old_block <- liftIO $ H.lookup ht global_stream_id - new_block <- case maybe_old_block of - - Nothing -> do - INSTRUMENTATION( liftIO $ infoM "HTTP2.Session" $ "Starting stream " ++ (show global_stream_id) ) - return $ Bu.byteString bytes - - Just something -> - return $ something `mappend` (Bu.byteString bytes) - - liftIO $ H.insert ht global_stream_id new_block - - -getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString -getHeaderBytes global_stream_id = do - ht <- view stream2HeaderBlockFragment - Just bytes <- liftIO $ H.lookup ht global_stream_id - return $ Bl.toStrict $ Bu.toLazyByteString bytes - - -frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString) -frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) ) - = Just (NH2.fromStreamIdentifier stream_id, block_fragment) -frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) ) - = Just (NH2.fromStreamIdentifier stream_id, block_fragment) -frameIsHeaderOfStream _ - = Nothing - - -frameEndsHeaders :: InputFrame -> Bool -frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags - - -streamIdFromFrame :: InputFrame -> GlobalStreamId -streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id - - --- TODO: Have different size for the headers..... just now going with a default size of 16 k... --- TODO: Find a way to kill this thread.... -headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers) - -> MVar (Chan (Either SessionOutputCommand OutputFrame)) - -> ReaderT SessionData IO () -headersOutputThread input_chan session_output_mvar = forever $ do - (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan - -- liftIO $ putStrLn $ "Output headers: " ++ (show headers) - - -- First encode the headers using the table - encode_dyn_table_mvar <- view toEncodeHeaders - - encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar - (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers - liftIO $ putMVar encode_dyn_table_mvar new_dyn_table - - -- Now split the bytestring in chunks of the needed size.... - bs_chunks <- return $! bytestringChunk useChunkLength data_to_send - - -- And send the chunks through while locking the output place.... - liftIO $ E.bracket - (takeMVar session_output_mvar) - (putMVar session_output_mvar ) - (\ session_output -> do - writeIndividualHeaderFrames session_output stream_id bs_chunks True - -- And say that the headers for this thread are out - INSTRUMENTATION( infoM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) ) - putMVar headers_ready_mvar HeadersSent - ) - where - writeIndividualHeaderFrames :: - Chan (Either SessionOutputCommand OutputFrame) - -> GlobalStreamId - -> [B.ByteString] - -> Bool - -> IO () - writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first = - writeChan session_output $ Right ( NH2.EncodeInfo { - NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, - (if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment) - ) - writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first = do - writeChan session_output $ Right ( NH2.EncodeInfo { - NH2.encodeFlags = NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, - (if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment) - ) - writeIndividualHeaderFrames session_output stream_id xs False - - -bytestringChunk :: Int -> B.ByteString -> [B.ByteString] -bytestringChunk len s | (B.length s) < len = [ s ] -bytestringChunk len s = h:(bytestringChunk len xs) - where - (h, xs) = B.splitAt len s - - --- TODO: find a clean way to finish this thread (maybe with negative stream ids?) --- TODO: This function does non-optimal chunking for the case where responses are --- actually streamed.... in those cases we need to keep state for frames in --- some other format.... --- TODO: Right now, we are transmitting an empty last frame with the end-of-stream --- flag set. I'm afraid that the only --- way to avoid that is by holding a frame or by augmenting the end-user interface --- so that the user can signal which one is the last frame. The first approach --- restricts responsiviness, the second one clutters things. -dataOutputThread :: Chan DataOutputToConveyor - -> MVar (Chan (Either SessionOutputCommand OutputFrame)) - -> IO () -dataOutputThread input_chan session_output_mvar = forever $ do - (stream_id, maybe_contents) <- readChan input_chan - case maybe_contents of - Nothing -> do - liftIO $ do - INSTRUMENTATION( debugM "HTTP2.Session" "End-of-stream flag set " ) - withLockedSessionOutput - (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo { - NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, - NH2.DataFrame "" - ) - ) - - Just contents -> do - -- And now just simply output it... - let bs_chunks = bytestringChunk useChunkLength $! contents - -- And send the chunks through while locking the output place.... - writeContinuations bs_chunks stream_id - - where - - withLockedSessionOutput = E.bracket - (takeMVar session_output_mvar) - (putMVar session_output_mvar) -- <-- There is an implicit argument there!! - - writeContinuations :: [B.ByteString] -> GlobalStreamId -> IO () - writeContinuations fragments stream_id = mapM_ (\ fragment -> - withLockedSessionOutput (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo { - NH2.encodeFlags = NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, - NH2.DataFrame fragment ) ) - ) fragments - - diff --git a/hs-src/SecondTransfer/Http2/Session.hs b/hs-src/SecondTransfer/Http2/Session.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Session.hs @@ -0,0 +1,965 @@ +-- Session: links frames to streams, and helps in ordering the header frames +-- so that they don't get mixed with header frames from other streams when +-- resources are being served concurrently. +{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-} +{-# OPTIONS_HADDOCK hide #-} +module SecondTransfer.Http2.Session( + http2Session + ,getFrameFromSession + ,sendFrameToSession + ,sendCommandToSession + + ,CoherentSession + ,SessionInput(..) + ,SessionInputCommand(..) + ,SessionOutput(..) + ,SessionOutputCommand(..) + ,SessionCoordinates(..) + ,SessionComponent(..) + ,SessionsCallbacks(..) + ,SessionsConfig(..) + ,ErrorCallback + + -- Internal stuff + ,OutputFrame + ,InputFrame + ) where + +#include "Logging.cpphs" + +-- System grade utilities +import Control.Concurrent (ThreadId, forkIO) +import Control.Concurrent.Chan +import Control.Exception (throwTo) +import qualified Control.Exception as E +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Reader + +import Control.Concurrent.MVar +import qualified Data.ByteString as B +import qualified Data.ByteString.Builder as Bu +import qualified Data.ByteString.Lazy as Bl +import Data.Conduit +import qualified Data.HashTable.IO as H +import qualified Data.IntSet as NS +import Data.Maybe (isJust) +#ifndef IMPLICIT_MONOID +import Data.Monoid (mappend) +#endif + +import Control.Lens + +-- No framing layer here... let's use Kazu's Yamamoto library +import qualified Network.HPACK as HP +import qualified Network.HTTP2 as NH2 + +-- Logging utilities +import System.Log.Logger + +-- Imports from other parts of the program +import SecondTransfer.MainLoop.CoherentWorker +import SecondTransfer.MainLoop.Tokens +import SecondTransfer.Sessions.Config +import SecondTransfer.Sessions.Internal (sessionExceptionHandler, SessionsContext, sessionsConfig) +import SecondTransfer.Utils (unfoldChannelAndSource) +import SecondTransfer.Exception +import qualified SecondTransfer.Utils.HTTPHeaders as He + +-- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to +-- use :-( +type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload) +type InputFrame = NH2.Frame + + +useChunkLength :: Int +useChunkLength = 16384 + + +-- Singleton instance used for concurrency +data HeadersSent = HeadersSent + +-- All streams put their data bits here. A "Nothing" value signals +-- end of data. +type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString) + + +-- Whatever a worker thread is going to need comes here.... +-- this is to make refactoring easier, but not strictly needed. +data WorkerThreadEnvironment = WorkerThreadEnvironment { + -- What's the header stream id? + _streamId :: GlobalStreamId + + -- A full block of headers can come here... the mvar in the middle should + -- be populate to signal end of headers transmission. A thread will be suspended + -- waiting for that + , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers) + + -- And regular contents can come this way and thus be properly mixed + -- with everything else.... for now... + ,_dataOutput :: Chan DataOutputToConveyor + + ,_streamsCancelled_WTE :: MVar NS.IntSet + + } + +makeLenses ''WorkerThreadEnvironment + + +-- An HTTP/2 session. Basically a couple of channels ... +type Session = (SessionInput, SessionOutput) + + +-- From outside, one can only write to this one ... the newtype is to enforce +-- this. +newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) ) +sendFrameToSession :: SessionInput -> InputFrame -> IO () +sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame + +sendCommandToSession :: SessionInput -> SessionInputCommand -> IO () +sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command + +-- From outside, one can only read from this one +newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) ) +getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) +getFrameFromSession (SessionOutput chan) = readChan chan + + +type HashTable k v = H.CuckooHashTable k v + + +type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder + + +type WorkerMonad = ReaderT WorkerThreadEnvironment IO + + +-- Have to figure out which are these...but I would expect to have things +-- like unexpected aborts here in this type. +data SessionInputCommand = + CancelSession_SIC + deriving Show + + +-- temporary +data SessionOutputCommand = + CancelSession_SOC + deriving Show + + +-- Here is how we make a session +type SessionMaker = SessionsContext -> IO Session + + +-- Here is how we make a session wrapping a CoherentWorker +type CoherentSession = CoherentWorker -> SessionMaker + + +data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream) + +-- Settings imposed by the peer +data SessionSettings = SessionSettings { + _pushEnabled :: Bool + } + +makeLenses ''SessionSettings + + +-- NH2.Frame != Frame +data SessionData = SessionData { + -- ATTENTION: Ignore the warning coming from here for now + _sessionsContext :: SessionsContext + + ,_sessionInput :: Chan (Either SessionInputCommand InputFrame) + + -- We need to lock this channel occassionally so that we can order multiple + -- header frames properly.... + ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame)) + + -- Use to encode + ,_toEncodeHeaders :: MVar HP.DynamicTable + -- And used to decode + ,_toDecodeHeaders :: MVar HP.DynamicTable + -- While I'm receiving headers, anything which + -- is not a header should end in the connection being + -- closed + ,_receivingHeaders :: MVar (Maybe Int) + -- _lastGoodStream is used both to report the last good stream in the + -- GoAwayFrame and to keep track of streams oppened by the client. In + -- other words, it contains the stream_id of the last valid client + -- stream and is updated as soon as the first frame of that stream is + -- received. + ,_lastGoodStream :: MVar Int + + -- Used for decoding the headers + ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment + + -- Used for worker threads... this is actually a pre-filled template + -- I make copies of it in different contexts, and as needed. + ,_forWorkerThread :: WorkerThreadEnvironment + + ,_coherentWorker :: CoherentWorker + + -- Some streams may be cancelled + ,_streamsCancelled :: MVar NS.IntSet + + -- Data input mechanism corresponding to some threads + ,_stream2PostInputMechanism :: HashTable Int PostInputMechanism + + -- Worker thread register. This is a dictionary from stream id to + -- the ThreadId of the thread with the worker thread. I use this to + -- raise asynchronous exceptions in the worker thread if the stream + -- is cancelled by the client. This way we get early finalization. + ,_stream2WorkerThread :: HashTable Int ThreadId + + -- Use to retrieve/set the session id + ,_sessionIdAtSession :: Int + + -- And used to keep peer session settings + ,_sessionSettings :: MVar SessionSettings + } + + +makeLenses ''SessionData + + +-- v- {headers table size comes here!!} +http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session +http2Session coherent_worker session_id sessions_context = do + session_input <- newChan + session_output <- newChan + session_output_mvar <- newMVar session_output + + + -- For incremental construction of headers... + stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment + + -- Warning: we should find a way of coping with different table sizes. + decode_headers_table <- HP.newDynamicTableForDecoding 4096 + decode_headers_table_mvar <- newMVar decode_headers_table + + encode_headers_table <- HP.newDynamicTableForEncoding 4096 + encode_headers_table_mvar <- newMVar encode_headers_table + + -- These ones need independent threads taking care of sending stuff + -- their way... + headers_output <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers)) + data_output <- newChan :: IO (Chan DataOutputToConveyor) + + stream2postinputmechanism <- H.new + stream2workerthread <- H.new + last_good_stream_mvar <- newMVar (-1) + + receiving_headers <- newMVar Nothing + session_settings <- newMVar $ SessionSettings { _pushEnabled = True } + + -- What about stream cancellation? + cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet) + + let for_worker_thread = WorkerThreadEnvironment { + _streamId = error "NotInitialized" + ,_headersOutput = headers_output + ,_dataOutput = data_output + ,_streamsCancelled_WTE = cancelled_streams_mvar + } + + let session_data = SessionData { + _sessionsContext = sessions_context + ,_sessionInput = session_input + ,_sessionOutput = session_output_mvar + ,_toDecodeHeaders = decode_headers_table_mvar + ,_toEncodeHeaders = encode_headers_table_mvar + ,_stream2HeaderBlockFragment = stream_request_headers + ,_forWorkerThread = for_worker_thread + ,_coherentWorker = coherent_worker + ,_streamsCancelled = cancelled_streams_mvar + ,_stream2PostInputMechanism = stream2postinputmechanism + ,_stream2WorkerThread = stream2workerthread + ,_sessionIdAtSession = session_id + ,_receivingHeaders = receiving_headers + ,_sessionSettings = session_settings + ,_lastGoodStream = last_good_stream_mvar + } + + let + exc_handler :: SessionComponent -> HTTP2SessionException -> IO () + exc_handler component e = sessionExceptionHandler component session_id sessions_context e + exc_guard :: SessionComponent -> IO () -> IO () + exc_guard component action = E.catch + action + (\e -> do + INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" ) + exc_handler component e + ) + + -- Create an input thread that decodes frames... + forkIO $ exc_guard SessionInputThread_HTTP2SessionComponent + $ runReaderT sessionInputThread session_data + + -- Create a thread that captures headers and sends them down the tube + forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent + $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data + + -- Create a thread that captures data and sends it down the tube + forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent + $ dataOutputThread data_output session_output_mvar + + -- The two previous threads fill the session_output argument below (they write to it) + -- the session machinery in the other end is in charge of sending that data through the + -- socket. + + return ( (SessionInput session_input), + (SessionOutput session_output) ) + + +-- TODO: Some ill clients can break this thread with exceptions. Make these paths a bit +--- more robust. +sessionInputThread :: ReaderT SessionData IO () +sessionInputThread = do + INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread" ) + + -- This is an introductory and declarative block... all of this is tail-executed + -- every time that a packet needs to be processed. It may be a good idea to abstract + -- these values in a closure... + session_input <- view sessionInput + + decode_headers_table_mvar <- view toDecodeHeaders + stream_request_headers <- view stream2HeaderBlockFragment + cancelled_streams_mvar <- view streamsCancelled + coherent_worker <- view coherentWorker + + for_worker_thread_uns <- view forWorkerThread + stream2workerthread <- view stream2WorkerThread + receiving_headers_mvar <- view receivingHeaders + last_good_stream_mvar <- view lastGoodStream + + input <- liftIO $ readChan session_input + + -- INSTRUMENTATION( liftIO $ infoM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input) ) + + case input of + + Left CancelSession_SIC -> do + -- Good place to tear down worker threads... Let the rest of the finalization + -- to the framer + liftIO $ do + H.mapM_ + (\ (_, thread_id) -> do + throwTo thread_id StreamCancelledException + return () + ) + stream2workerthread + + -- We do not continue here, but instead let it finish + return () + + -- The block below will process both HEADERS and CONTINUATION frames. + -- TODO: As it stands now, the server will happily start a new stream with + -- a CONTINUATION frame instead of a HEADERS frame. That's against the + -- protocol. + Right frame | Just (stream_id, bytes) <- isAboutHeaders frame -> do + -- Just append the frames to streamRequestHeaders + opens_stream <- appendHeaderFragmentBlock stream_id bytes + + if opens_stream + then do + maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar + case maybe_rcv_headers_of of + Just _ -> do + -- INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" "headers being received") + -- Bad client, it is already sending headers + -- and trying to open another one + closeConnectionBecauseIsInvalid NH2.ProtocolError + -- An exception will be thrown above, so to not complicate + -- control flow here too much. + Nothing -> do + -- Signal that we are receiving headers now, for this stream + liftIO $ putMVar receiving_headers_mvar (Just stream_id) + -- And go to check if the stream id is valid + last_good_stream <- liftIO $ takeMVar last_good_stream_mvar + if (odd stream_id ) && (stream_id > last_good_stream) + then do + -- We are golden, set the new good stream + liftIO $ putMVar last_good_stream_mvar (stream_id) + else do + -- We are not golden + INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" "Protocol error: bad stream id") + closeConnectionBecauseIsInvalid NH2.ProtocolError + else do + maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar + case maybe_rcv_headers_of of + Just a_stream_id | a_stream_id == stream_id -> do + -- Nothing to complain about + liftIO $ putMVar receiving_headers_mvar maybe_rcv_headers_of + + Nothing -> error "InternalError, this should be set" + + if frameEndsHeaders frame then + do + -- Ok, let it be known that we are not receiving more headers + liftIO $ modifyMVar_ + receiving_headers_mvar + (\ _ -> return Nothing ) + -- Let's decode the headers + let for_worker_thread = set streamId stream_id for_worker_thread_uns + headers_bytes <- getHeaderBytes stream_id + dyn_table <- liftIO $ takeMVar decode_headers_table_mvar + (new_table, header_list ) <- liftIO $ HP.decodeHeader dyn_table headers_bytes + -- Good moment to remove the headers from the table.... we don't want a space + -- leak here + liftIO $ do + H.delete stream_request_headers stream_id + putMVar decode_headers_table_mvar new_table + + -- TODO: Validate headers, abort session if the headers are invalid. + -- Otherwise other invariants will break!! + -- THIS IS PROBABLY THE BEST PLACE FOR DOING IT. + let + headers_editor = He.fromList header_list + + maybe_good_headers_editor <- validateIncomingHeaders headers_editor + + good_headers <- case maybe_good_headers_editor of + Just yes_they_are_good -> return yes_they_are_good + Nothing -> closeConnectionBecauseIsInvalid NH2.ProtocolError + + -- Add any extra headers, on demand + headers_extra_good <- addExtraHeaders good_headers + let + header_list_after = He.toList headers_extra_good + -- liftIO $ putStrLn $ "header list after " ++ (show header_list_after) + + -- If the headers end the request.... + post_data_source <- if not (frameEndsStream frame) + then do + mechanism <- createMechanismForStream stream_id + let source = postDataSourceFromMechanism mechanism + return $ Just source + else do + return Nothing + + -- TODO: Handle the cases where a request tries to send data + -- even if the method doesn't allow for data. + + -- I'm clear to start the worker, in its own thread + liftIO $ do + thread_id <- forkIO $ runReaderT + (workerThread (header_list_after, post_data_source) coherent_worker) + for_worker_thread + H.insert stream2workerthread stream_id thread_id + + return () + else + -- Frame doesn't end the headers... it was added before... so + -- probably do nothing + return () + + continue + + Right frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do + let stream_id = streamIdFromFrame frame + liftIO $ do + INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show _error_code_id) ) + cancelled_streams <- takeMVar cancelled_streams_mvar + INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) ) + putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams + maybe_thread_id <- H.lookup stream2workerthread stream_id + case maybe_thread_id of + Nothing -> + -- This is actually more like an internal error, when this + -- happend, cancell the session + error "InterruptingUnexistentStream" + + Just thread_id -> do + INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" ) + throwTo thread_id StreamCancelledException + + continue + + Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) + -> unlessReceivingHeaders $ do + -- So I got data to process + -- TODO: Handle end of stream + let stream_id = NH2.fromStreamIdentifier nh2_stream_id + -- TODO: Handle the cases where the stream_id doesn't match an already existent + -- stream. In such cases it is justified to reset the connection with a protocol_error. + + streamWorkerSendData stream_id somebytes + -- After that data has been received and forwarded downstream, we can issue a windows update + -- + -- TODO: We can use wider frames to avoid congestion... + -- .... and we can also be more compositional with these short bursts of data.... + -- + -- TODO: Consider that the best place to output these frames can be somewhere else... + -- + -- TODO: Use a special, with-quota queue here to do flow control. Don't send meaningless + -- WindowUpdateFrame's + sendOutFrame + (NH2.EncodeInfo + NH2.defaultFlags + nh2_stream_id + Nothing + ) + (NH2.WindowUpdateFrame + (fromIntegral (B.length somebytes)) + ) + sendOutFrame + (NH2.EncodeInfo + NH2.defaultFlags + (NH2.toStreamIdentifier 0) + Nothing + ) + (NH2.WindowUpdateFrame + (fromIntegral (B.length somebytes)) + ) + + if frameEndsStream frame + then do + -- Good place to close the source ... + closePostDataSource stream_id + else + return () + + continue + + Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do + -- Deal with pings: this is an Ack, so do nothing + continue + + Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do + -- Deal with pings: NOT an Ack, so answer + INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Ping processed" ) + sendOutFrame + (NH2.EncodeInfo + (NH2.setAck NH2.defaultFlags) + (NH2.toStreamIdentifier 0) + Nothing + ) + (NH2.PingFrame somebytes) + + continue + + Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do + -- Frame was received by the peer, do nothing here... + continue + + -- TODO: Do something with these settings!! + Right (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do + INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) ) + -- Just acknowledge the frame.... for now + sendOutFrame + (NH2.EncodeInfo + (NH2.setAck NH2.defaultFlags) + (NH2.toStreamIdentifier 0) + Nothing ) + (NH2.SettingsFrame []) + + continue + + + Right somethingelse -> unlessReceivingHeaders $ do + -- An undhandled case here.... + INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: " ) + liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse) + + continue + + where + continue = sessionInputThread + + +sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO () +sendOutFrame encode_info payload = do + session_output_mvar <- view sessionOutput + + session_output <- liftIO $ takeMVar session_output_mvar + liftIO $ writeChan session_output $ Right (encode_info, payload) + liftIO $ putMVar session_output_mvar session_output + + +-- TODO: This function, but using the headers editor, triggers +-- some renormalization of the header order. A good thing, if +-- I get that order well enough.... +addExtraHeaders :: He.HeaderEditor -> ReaderT SessionData IO He.HeaderEditor +addExtraHeaders headers_editor = do + let + enriched_lens = (sessionsContext . sessionsConfig .sessionsEnrichedHeaders ) + -- TODO: Figure out which is the best way to put this contact in the + -- source code + protocol_lens = He.headerLens "second-transfer-eh--used-protocol" + + add_used_protocol <- view (enriched_lens . addUsedProtocol ) + + -- liftIO $ putStrLn $ "AAA" ++ (show add_used_protocol) + + let + he1 = if add_used_protocol + then set protocol_lens (Just "HTTP/2") headers_editor + else headers_editor + + if add_used_protocol + -- Nothing will be computed here if the headers are not modified. + then return he1 + else return headers_editor + + +validateIncomingHeaders :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor) +validateIncomingHeaders headers_editor = do + -- Check that the headers block comes with all mandatory headers. + -- Right now I'm not checking that they come in the mandatory order though... + -- + -- Notice that this function will transform a "host" header to an ":authority" + -- one. + let + h1 = He.replaceHostByAuthority headers_editor + -- Check that headers are lowercase + headers_are_lowercase = He.headersAreLowercaseAtHeaderEditor headers_editor + -- Check that we have mandatory headers + maybe_authority = h1 ^. (He.headerLens ":authority") + maybe_method = h1 ^. (He.headerLens ":method") + maybe_scheme = h1 ^. (He.headerLens ":scheme") + maybe_path = h1 ^. (He.headerLens ":path") + + if + (isJust maybe_authority) && + (isJust maybe_method) && + (isJust maybe_scheme) && + (isJust maybe_path ) + then + return (Just h1) + else + return Nothing + + +-- Sends a GO_AWAY frame and raises an exception, effectively terminating the input +-- thread of the session. +closeConnectionBecauseIsInvalid :: NH2.ErrorCodeId -> ReaderT SessionData IO a +closeConnectionBecauseIsInvalid error_code = do + liftIO $ errorM "HTTP2.Session" "closeConnectionBecauseIsInvalid called!" + last_good_stream_mvar <- view lastGoodStream + last_good_stream <- liftIO $ takeMVar last_good_stream_mvar + session_output_mvar <- view sessionOutput + stream2workerthread <- view stream2WorkerThread + sendOutFrame + (NH2.EncodeInfo + NH2.defaultFlags + (NH2.toStreamIdentifier 0) + Nothing + ) + (NH2.GoAwayFrame + (NH2.toStreamIdentifier last_good_stream) + error_code + "" + ) + + liftIO $ do + -- Close all active threads for this session + H.mapM_ + ( \(_stream_id, thread_id) -> + throwTo thread_id StreamCancelledException + ) + stream2workerthread + + -- Notify the framer that the session is closing, so + -- that it stops accepting frames from connected sources + -- (Streams?) + session_output <- takeMVar session_output_mvar + writeChan session_output $ Left CancelSession_SOC + putMVar session_output_mvar session_output + + -- And unwind the input thread in the session, so that the + -- exception handler runs.... + E.throw HTTP2ProtocolException + + +frameEndsStream :: InputFrame -> Bool +frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags + + +-- Executes its argument, unless receiving +-- headers, in which case the connection is closed. +unlessReceivingHeaders :: ReaderT SessionData IO a -> ReaderT SessionData IO a +unlessReceivingHeaders comp = do + receiving_headers_mvar <- view receivingHeaders + -- First check if we are receiving headers + maybe_recv_headers <- liftIO $ readMVar receiving_headers_mvar + if isJust maybe_recv_headers + then + -- So, this frame is highly illegal + closeConnectionBecauseIsInvalid NH2.ProtocolError + else + comp + + +createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism +createMechanismForStream stream_id = do + (chan, source) <- liftIO $ unfoldChannelAndSource + stream2postinputmechanism <- view stream2PostInputMechanism + let pim = PostInputMechanism (chan, source) + liftIO $ H.insert stream2postinputmechanism stream_id pim + return pim + + +-- TODO: Can be optimized by factoring out the mechanism lookup +-- TODO IMPORTANT: This is a good place to drop the postinputmechanism +-- for a stream, so that unprocessed data can be garbage-collected. +closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO () +closePostDataSource stream_id = do + stream2postinputmechanism <- view stream2PostInputMechanism + + pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id + + case pim_maybe of + + Just (PostInputMechanism (chan, _)) -> + liftIO $ writeChan chan Nothing + + Nothing -> + -- TODO: This is a protocol error, handle it properly + error "Internal error/closePostDataSource" + + +streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO () +streamWorkerSendData stream_id bytes = do + s2pim <- view stream2PostInputMechanism + pim_maybe <- liftIO $ H.lookup s2pim stream_id + + case pim_maybe of + + Just pim -> + sendBytesToPim pim bytes + + Nothing -> + -- This is an internal error, the mechanism should be + -- created when the headers end (and if the headers + -- do not finish the stream) + error "Internal error" + + +sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO () +sendBytesToPim (PostInputMechanism (chan, _)) bytes = + liftIO $ writeChan chan (Just bytes) + + +postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream +postDataSourceFromMechanism (PostInputMechanism (_, source)) = source + + +isSettingsAck :: NH2.FrameHeader -> Bool +isSettingsAck (NH2.FrameHeader _ flags _) = + NH2.testAck flags + + +isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool +isStreamCancelled stream_id = do + cancelled_streams_mvar <- view streamsCancelled_WTE + cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar + return $ NS.member stream_id cancelled_streams + + + +workerThread :: Request -> CoherentWorker -> WorkerMonad () +workerThread req coherent_worker = + do + headers_output <- view headersOutput + stream_id <- view streamId + + -- TODO: Handle exceptions here: what happens if the coherent worker + -- throws an exception signaling that the request is ill-formed + -- and should be dropped? That could happen in a couple of occassions, + -- but really most cases should be handled here in this file... + (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req + + -- Now I send the headers, if that's possible at all + headers_sent <- liftIO $ newEmptyMVar + liftIO $ writeChan headers_output (stream_id, headers_sent, headers) + + -- At this moment I should ask if the stream hasn't been cancelled by the browser before + -- commiting to the work of sending addtitional data + is_stream_cancelled <- isStreamCancelled stream_id + if not is_stream_cancelled + + then do + -- I have a beautiful source that I can de-construct... + -- TODO: Optionally pulling data out from a Conduit .... + -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) ) + -- + -- This threadlet should block here waiting for the headers to finish going + (maybe_footers, _) <- runConduit $ + (transPipe liftIO data_and_conclussion) + `fuseBothMaybe` + (sendDataOfStream stream_id headers_sent) + -- BIG TODO: Send the footers ... likely stream conclusion semantics + -- will need to be changed. + return () + else + + return () + +-- v-- comp. monad. +sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) () +sendDataOfStream stream_id headers_sent = do + data_output <- view dataOutput + -- Wait for all headers sent + liftIO $ takeMVar headers_sent + consumer data_output + where + consumer data_output = do + maybe_bytes <- await + case maybe_bytes of + Nothing -> + liftIO $ writeChan data_output (stream_id, Nothing) + Just bytes -> do + liftIO $ writeChan data_output (stream_id, Just bytes) + consumer data_output + + +-- Returns if the frame is the first in the stream +appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO Bool +appendHeaderFragmentBlock global_stream_id bytes = do + ht <- view stream2HeaderBlockFragment + maybe_old_block <- liftIO $ H.lookup ht global_stream_id + (new_block, new_stream) <- case maybe_old_block of + + Nothing -> do + -- TODO: Make the commented message below more informative + -- INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Starting stream " ++ (show global_stream_id) ) + return $ (Bu.byteString bytes, True) + + Just something -> + return $ (something `mappend` (Bu.byteString bytes), False) + + liftIO $ H.insert ht global_stream_id new_block + return new_stream + +getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString +getHeaderBytes global_stream_id = do + ht <- view stream2HeaderBlockFragment + Just bytes <- liftIO $ H.lookup ht global_stream_id + return $ Bl.toStrict $ Bu.toLazyByteString bytes + + +isAboutHeaders :: InputFrame -> Maybe (GlobalStreamId, B.ByteString) +isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) ) + = Just (NH2.fromStreamIdentifier stream_id, block_fragment) +isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) ) + = Just (NH2.fromStreamIdentifier stream_id, block_fragment) +isAboutHeaders _ + = Nothing + + +frameEndsHeaders :: InputFrame -> Bool +frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags + + +streamIdFromFrame :: InputFrame -> GlobalStreamId +streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id + + +-- TODO: Have different size for the headers..... just now going with a default size of 16 k... +-- TODO: Find a way to kill this thread.... +headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers) + -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> ReaderT SessionData IO () +headersOutputThread input_chan session_output_mvar = forever $ do + (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan + + -- First encode the headers using the table + encode_dyn_table_mvar <- view toEncodeHeaders + + encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar + (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers + liftIO $ putMVar encode_dyn_table_mvar new_dyn_table + + -- Now split the bytestring in chunks of the needed size.... + bs_chunks <- return $! bytestringChunk useChunkLength data_to_send + + -- And send the chunks through while locking the output place.... + liftIO $ E.bracket + (takeMVar session_output_mvar) + (putMVar session_output_mvar ) + (\ session_output -> do + writeIndividualHeaderFrames session_output stream_id bs_chunks True + -- And say that the headers for this thread are out + -- INSTRUMENTATION( debugM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) ) + putMVar headers_ready_mvar HeadersSent + ) + where + writeIndividualHeaderFrames :: + Chan (Either SessionOutputCommand OutputFrame) + -> GlobalStreamId + -> [B.ByteString] + -> Bool + -> IO () + writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first = + writeChan session_output $ Right ( NH2.EncodeInfo { + NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + (if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment) + ) + writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first = do + writeChan session_output $ Right ( NH2.EncodeInfo { + NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + (if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment) + ) + writeIndividualHeaderFrames session_output stream_id xs False + + +bytestringChunk :: Int -> B.ByteString -> [B.ByteString] +bytestringChunk len s | (B.length s) < len = [ s ] +bytestringChunk len s = h:(bytestringChunk len xs) + where + (h, xs) = B.splitAt len s + + +-- TODO: find a clean way to finish this thread (maybe with negative stream ids?) +-- TODO: This function does non-optimal chunking for the case where responses are +-- actually streamed.... in those cases we need to keep state for frames in +-- some other format.... +-- TODO: Right now, we are transmitting an empty last frame with the end-of-stream +-- flag set. I'm afraid that the only +-- way to avoid that is by holding a frame or by augmenting the end-user interface +-- so that the user can signal which one is the last frame. The first approach +-- restricts responsiviness, the second one clutters things. +dataOutputThread :: Chan DataOutputToConveyor + -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> IO () +dataOutputThread input_chan session_output_mvar = forever $ do + (stream_id, maybe_contents) <- readChan input_chan + case maybe_contents of + Nothing -> do + liftIO $ do + withLockedSessionOutput + (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo { + NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + NH2.DataFrame "" + ) + ) + + Just contents -> do + -- And now just simply output it... + let bs_chunks = bytestringChunk useChunkLength $! contents + -- And send the chunks through while locking the output place.... + writeContinuations bs_chunks stream_id + + where + + withLockedSessionOutput = E.bracket + (takeMVar session_output_mvar) + (putMVar session_output_mvar) -- <-- There is an implicit argument there!! + + writeContinuations :: [B.ByteString] -> GlobalStreamId -> IO () + writeContinuations fragments stream_id = mapM_ (\ fragment -> + withLockedSessionOutput (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo { + NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + NH2.DataFrame fragment ) ) + ) fragments + + diff --git a/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs b/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs --- a/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs +++ b/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs @@ -8,6 +8,7 @@ module SecondTransfer.MainLoop.CoherentWorker( getHeaderFromFlatList + , nullFooter , HeaderName , HeaderValue @@ -22,6 +23,7 @@ , PushedStream , DataAndConclusion , InputDataStream + ) where @@ -80,6 +82,14 @@ -- all the headers of a request. For GET requests that's the entire request -- basically, but for POST and PUT requests this is just before the data -- starts arriving to the server. +-- +-- It is important that you consume the data in the cases where there is an +-- input stream, otherwise the memory is lost for the duration of the request, +-- and a malicious client can use that. +-- +-- Also, notice that when handling requests your worker can be interrupted with +-- an asynchronous exception of type 'StreamCancelledException', if the peer +-- cancels the stream type CoherentWorker = Request -> IO PrincipalStream @@ -103,3 +113,18 @@ Nothing -> Nothing + +-- | If you want to skip the footers, i.e., they are empty, use this +-- function to convert an ordinary Source to a DataAndConclusion. +nullFooter :: Source IO B.ByteString -> DataAndConclusion +nullFooter s = s =$= go + where + go = do + i <- await + case i of + Nothing -> + return [] + + Just ii -> do + yield ii + go diff --git a/hs-src/SecondTransfer/MainLoop/Framer.hs b/hs-src/SecondTransfer/MainLoop/Framer.hs --- a/hs-src/SecondTransfer/MainLoop/Framer.hs +++ b/hs-src/SecondTransfer/MainLoop/Framer.hs @@ -18,7 +18,10 @@ import qualified Data.ByteString.Lazy as LB import Data.Conduit --- import Data.Monoid (mappend, mempty) +#ifndef IMPLICIT_MONOID +import Data.Monoid +#endif + type Framer m = LB.ByteString -- Input left overs diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs deleted file mode 100644 --- a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs +++ /dev/null @@ -1,365 +0,0 @@ -{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings, DeriveDataTypeable #-} -{-# OPTIONS_HADDOCK hide #-} -module SecondTransfer.MainLoop.OpenSSL_TLS( - tlsServeWithALPN - ,tlsServeWithALPNAndFinishOnRequest - -- ,tlsServeWithALPNOnce - - ,TLSLayerGenericProblem(..) - ,FinishRequest(..) - ) where - - - -import Control.Monad -import Control.Concurrent.MVar -import Control.Exception -import qualified Control.Exception as E -import Data.Foldable (foldMap) -import Data.Typeable -import Data.Monoid () -import Foreign -import Foreign.C - -import qualified Data.ByteString as B -import qualified Data.ByteString.Builder as BB -import Data.ByteString.Char8 (pack) -import qualified Data.ByteString.Lazy as LB -import qualified Data.ByteString.Unsafe as BU - -import System.Log.Logger - -import SecondTransfer.MainLoop.PushPullType -import SecondTransfer.Exception - -#include "Logging.cpphs" - --- | Exceptions inheriting from `IOProblem`. This is thrown by the --- OpenSSL subsystem to signal that the connection was broken or that --- otherwise there was a problem at the SSL layer. -data TLSLayerGenericProblem = TLSLayerGenericProblem String - deriving (Show, Typeable) - - -instance Exception TLSLayerGenericProblem where - toException = toException . IOProblem - fromException x = do - IOProblem a <- fromException x - cast a - - -data InterruptibleEither a b = - Left_I a - |Right_I b - |Interrupted - - --- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, --- the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish --- at its earliest convenience and call the `CloseAction` for any open sessions. -data FinishRequest = FinishRequest - - --- These names are absolutely improper.... --- Session creator -data Connection_t --- Session -data Wired_t - -type Connection_Ptr = Ptr Connection_t -type Wired_Ptr = Ptr Wired_t - - --- Actually, this makes a listener for new connections --- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, --- char* protocol_list, int protocol_list_len) -foreign import ccall "make_connection" makeConnection :: - CString -- cert filename - -> CString -- privkey_filename - -> CString -- hostname - -> CInt -- port - -> Ptr CChar -- protocol list - -> CInt -- protocol list length - -> IO Connection_Ptr - -allOk :: CInt -allOk = 0 - -badHappened :: CInt -badHappened = 1 - -timeoutReached :: CInt -timeoutReached = 3 - --- int wait_for_connection(connection_t* conn, wired_session_t** wired_session); -foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt - --- int send_data(wired_session_t* ws, char* buffer, int buffer_size); -foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt - --- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd); -foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt - --- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; } -foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt - --- void dispose_wired_session(wired_session_t* ws); -foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO () - -foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO () - - -useBufferSize :: Int -useBufferSize = 4096 - - -type Protocols = [B.ByteString] - - -protocolsToWire :: Protocols -> B.ByteString -protocolsToWire protocols = - LB.toStrict . BB.toLazyByteString $ - foldMap (\ protocol - -> (BB.lazyByteString . LB.fromChunks) - [ B.singleton $ fromIntegral $ B.length protocol, - protocol - ] - ) protocols - - --- | Simple function to open -tlsServeWithALPN :: FilePath -- ^ Path to a certificate the server is going to use to identify itself. - -- Bear in mind that multiple domains can be served from the same HTTP/2 - -- TLS socket, so please create the HTTP/2 certificate accordingly. - -- Also, currently this function only accepts paths to certificates - -- or certificate chains in .pem format. - -> FilePath -- ^ Path to the key of your certificate. - -> String -- ^ Name of the network interface where you want to start your server - -> [(String, Attendant)] -- ^ List of protocol names and the corresponding `Attendant` to use for - -- each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the - -- same socket. When no ALPN negotiation is present during the negotiation, - -- the first protocol in this list is used. - -> Int -- ^ Port to open to listen for connections. - -> IO () -tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do - - let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants - INSTRUMENTATION( infoM "OpenSSL" "Entering tlsServeWithALPN" ) - withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do - - connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) -> - makeConnection - c_certfn - c_keyfn - c_iname - (fromIntegral interface_port) - pchar - (fromIntegral len) - - if connection_ptr == nullPtr - then do - INSTRUMENTATION( errorM "OpenSSL" "Could not create listening socket" ) - throwIO $ TLSLayerGenericProblem "Could not create listening end" - else do - INSTRUMENTATION( infoM "OpenSSL" "Listening socket created" ) - return () - - forever $ do - either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do - result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr - let - r = case result_code of - re | re == allOk -> do - p <- peek wired_ptr_ptr - INSTRUMENTATION( infoM "OpenSSL" "A connection was accepted" ) - return $ Right p - | re == timeoutReached -> tryOnce - | re == badHappened -> return $ Left ("A wait for connection failed" :: String) - r - in tryOnce - - case either_wired_ptr of - --- Disable a warning - Left _msg -> do - INSTRUMENTATION( errorM "OpenSSL" $ ".. wait for connection failed. " ++ _msg ) - return () - - Right wired_ptr -> do - already_closed_mvar <- newMVar False - let - pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do - result <- sendData wired_ptr pchar (fromIntegral len) - case result of - r | r == allOk -> return () - | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data" - pullAction = do - allocaBytes useBufferSize $ \ pcharbuffer -> - alloca $ \ data_recvd_ptr -> do - result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr - INSTRUMENTATION( debugM "OpenSSL" "Received data" ) - recvd_bytes <- case result of - r | r == allOk -> peek data_recvd_ptr - | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data" - - B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes) - - -- Ensure that the socket and the struct are only closed once - closeAction = do - -- debugM "OpenSSL" "About to close connection" - b <- readMVar already_closed_mvar - if not b - then do - modifyMVar_ already_closed_mvar (\ _ -> return True) - disposeWiredSession wired_ptr - -- debugM "OpenSSL" "dispose clalled" - else - return () - - use_protocol <- getSelectedProtocol wired_ptr - INSTRUMENTATION( infoM "OpenSSL" $ "Selected protocol:" ++ (show use_protocol)) - - let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n - -- Or just select the first one - | otherwise -> Just . snd . head $ attendants - - case maybe_session_attendant of - - Just session_attendant -> - E.catch - (session_attendant pushAction pullAction closeAction) - ((\ e -> do - INSTRUMENTATION( errorM "OpenSSL" " ** Session ended by TLSLayerGenericProblem (well handled)") - throwIO e - )::TLSLayerGenericProblem -> IO () ) - - - Nothing -> - return () - - --- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to ask --- the server to finish: you pass an empty MVar and when you want to finish you --- just populate it. -tlsServeWithALPNAndFinishOnRequest :: FilePath - -> FilePath -- ^ Same as for `tlsServeWithALPN` - -> String -- ^ Same as for `tlsServeWithALPN` - -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN` - -> Int -- ^ Same as for `tlsServeWithALPN` - -> MVar FinishRequest -- ^ Finish request, write a value here to finish serving - -> IO () -tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do - - let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants - withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do - - -- Create an accepting endpoint - connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) -> - makeConnection - c_certfn - c_keyfn - c_iname - (fromIntegral interface_port) - pchar - (fromIntegral len) - - -- Create a computation that accepts a connection, runs a session on it and recurses - let - recursion = do - -- Get a SSL session - either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do - result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr - let - r = case result_code of - re | re == allOk -> do - p <- peek wired_ptr_ptr - return $ Right_I p - | re == timeoutReached -> do - got_finish_request <- tryTakeMVar finish_request - case got_finish_request of - Nothing -> - tryOnce - Just _ -> - return Interrupted - - | re == badHappened -> return $ Left_I "A wait for connection failed" - r - in tryOnce - - -- With the potentially obtained SSL session do... - case either_wired_ptr of - - Left_I msg -> do - errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg - - -- // .. // - recursion - - Right_I wired_ptr -> do - already_closed_mvar <- newMVar False - let - pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do - result <- sendData wired_ptr pchar (fromIntegral len) - case result of - r | r == allOk -> return () - | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data" - pullAction = do - allocaBytes useBufferSize $ \ pcharbuffer -> - alloca $ \ data_recvd_ptr -> do - result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr - recvd_bytes <- case result of - r | r == allOk -> peek data_recvd_ptr - | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data" - - B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes) - closeAction = do - -- debugM "OpenSSL" "Close action about to be invoked" - b <- readMVar already_closed_mvar - if not b - then do - modifyMVar_ already_closed_mvar (\ _ -> return True) - disposeWiredSession wired_ptr - -- debugM "OpenSSL" "Close action invoked" - else - return () - - use_protocol <- getSelectedProtocol wired_ptr - - infoM "OpenSSL" $ ".. Using protocol: " ++ (show use_protocol) - - let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n - | otherwise -> Just . snd . head $ attendants - - case maybe_session_attendant of - - Just session_attendant -> - session_attendant pushAction pullAction closeAction - - Nothing -> - return () - - -- // .. // - recursion - - Interrupted -> do - infoM "OpenSSL" "Connection closed" - closeConnection connection_ptr - - -- Start the loop defined above... - recursion - --- When we are using the eternal version of this function, wake up --- each second .... -defaultWaitTime :: CInt -defaultWaitTime = 200000 --- Okej, more responsiviness needed -smallWaitTime :: CInt -smallWaitTime = 50000 diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs @@ -0,0 +1,350 @@ +{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings, DeriveDataTypeable #-} +{-# OPTIONS_HADDOCK hide #-} +module SecondTransfer.MainLoop.OpenSSL_TLS( + tlsServeWithALPN + ,tlsServeWithALPNAndFinishOnRequest + -- ,tlsServeWithALPNOnce + + ,TLSLayerGenericProblem(..) + ,FinishRequest(..) + ) where + +import Control.Monad +import Control.Concurrent.MVar +import Control.Exception +import qualified Control.Exception as E +#ifndef IMPLICIT_APPLICATIVE_FOLDABLE +import Data.Foldable (foldMap) +#endif +import Data.Typeable +import Data.Monoid () +import Foreign +import Foreign.C + +import qualified Data.ByteString as B +import qualified Data.ByteString.Builder as BB +import Data.ByteString.Char8 (pack) +import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Unsafe as BU + +-- import System.Log.Logger + +import SecondTransfer.MainLoop.PushPullType +import SecondTransfer.Exception + +#include "Logging.cpphs" + +-- | Exceptions inheriting from `IOProblem`. This is thrown by the +-- OpenSSL subsystem to signal that the connection was broken or that +-- otherwise there was a problem at the SSL layer. +data TLSLayerGenericProblem = TLSLayerGenericProblem String + deriving (Show, Typeable) + + +instance Exception TLSLayerGenericProblem where + toException = toException . IOProblem + fromException x = do + IOProblem a <- fromException x + cast a + + +data InterruptibleEither a b = + Left_I a + |Right_I b + |Interrupted + + +-- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, +-- the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish +-- at its earliest convenience and call the `CloseAction` for any open sessions. +data FinishRequest = FinishRequest + + +-- These names are absolutely improper.... +-- Session creator +data Connection_t +-- Session +data Wired_t + +type Connection_Ptr = Ptr Connection_t +type Wired_Ptr = Ptr Wired_t + + +-- Actually, this makes a listener for new connections +-- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, +-- char* protocol_list, int protocol_list_len) +foreign import ccall "make_connection" makeConnection :: + CString -- cert filename + -> CString -- privkey_filename + -> CString -- hostname + -> CInt -- port + -> Ptr CChar -- protocol list + -> CInt -- protocol list length + -> IO Connection_Ptr + +allOk :: CInt +allOk = 0 + +badHappened :: CInt +badHappened = 1 + +timeoutReached :: CInt +timeoutReached = 3 + +-- int wait_for_connection(connection_t* conn, wired_session_t** wired_session); +foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt + +-- int send_data(wired_session_t* ws, char* buffer, int buffer_size); +foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt + +-- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd); +foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt + +-- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; } +foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt + +-- void dispose_wired_session(wired_session_t* ws); +foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO () + +foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO () + + +useBufferSize :: Int +useBufferSize = 4096 + + +type Protocols = [B.ByteString] + + +protocolsToWire :: Protocols -> B.ByteString +protocolsToWire protocols = + LB.toStrict . BB.toLazyByteString $ + foldMap (\ protocol + -> (BB.lazyByteString . LB.fromChunks) + [ B.singleton $ fromIntegral $ B.length protocol, + protocol + ] + ) protocols + + +-- | Simple function to open +tlsServeWithALPN :: FilePath -- ^ Path to a certificate the server is going to use to identify itself. + -- Bear in mind that multiple domains can be served from the same HTTP/2 + -- TLS socket, so please create the HTTP/2 certificate accordingly. + -- Also, currently this function only accepts paths to certificates + -- or certificate chains in .pem format. + -> FilePath -- ^ Path to the key of your certificate. + -> String -- ^ Name of the network interface where you want to start your server + -> [(String, Attendant)] -- ^ List of protocol names and the corresponding `Attendant` to use for + -- each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the + -- same socket. When no ALPN negotiation is present during the negotiation, + -- the first protocol in this list is used. + -> Int -- ^ Port to open to listen for connections. + -> IO () +tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do + + let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants + withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do + + connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) -> + makeConnection + c_certfn + c_keyfn + c_iname + (fromIntegral interface_port) + pchar + (fromIntegral len) + + if connection_ptr == nullPtr + then do + throwIO $ TLSLayerGenericProblem "Could not create listening end" + else do + return () + + forever $ do + either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr + let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr + return $ Right p + | re == timeoutReached -> tryOnce + | re == badHappened -> return $ Left ("A wait for connection failed" :: String) + r + in tryOnce + + case either_wired_ptr of + +-- Disable a warning + Left _msg -> do + return () + + Right wired_ptr -> do + + (push_action, pull_action, close_action) <- provideActions wired_ptr + + use_protocol <- getSelectedProtocol wired_ptr + + let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + -- Or just select the first one + | otherwise -> Just . snd . head $ attendants + + case maybe_session_attendant of + + Just session_attendant -> + E.catch + (session_attendant push_action pull_action close_action) + ((\ e -> do + throwIO e + )::TLSLayerGenericProblem -> IO () ) + + + Nothing -> + return () + + +-- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to ask +-- the server to finish: you pass an empty MVar and when you want to finish you +-- just populate it. +tlsServeWithALPNAndFinishOnRequest :: FilePath + -> FilePath -- ^ Same as for `tlsServeWithALPN` + -> String -- ^ Same as for `tlsServeWithALPN` + -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN` + -> Int -- ^ Same as for `tlsServeWithALPN` + -> MVar FinishRequest -- ^ Finish request, write a value here to finish serving + -> IO () +tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do + + let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants + withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do + + -- Create an accepting endpoint + connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) -> + makeConnection + c_certfn + c_keyfn + c_iname + (fromIntegral interface_port) + pchar + (fromIntegral len) + + -- Create a computation that accepts a connection, runs a session on it and recurses + let + recursion = do + -- Get a SSL session + either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr + let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr + return $ Right_I p + | re == timeoutReached -> do + got_finish_request <- tryTakeMVar finish_request + case got_finish_request of + Nothing -> + tryOnce + Just _ -> + return Interrupted + + | re == badHappened -> return $ Left_I "A wait for connection failed" + r + in tryOnce + + -- With the potentially obtained SSL session do... + case either_wired_ptr of + + Left_I _msg -> do + -- // .. // + recursion + + Right_I wired_ptr -> do + (push_action, pull_action, close_action) <- provideActions wired_ptr + + use_protocol <- getSelectedProtocol wired_ptr + + let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + | otherwise -> Just . snd . head $ attendants + + case maybe_session_attendant of + + Just session_attendant -> + session_attendant push_action pull_action close_action + + Nothing -> + return () + + -- // .. // + recursion + + Interrupted -> do + closeConnection connection_ptr + + -- Start the loop defined above... + recursion + +-- When we are using the eternal version of this function, wake up +-- each second .... +defaultWaitTime :: CInt +defaultWaitTime = 200000 +-- Okej, more responsiviness needed +smallWaitTime :: CInt +smallWaitTime = 50000 + +provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), IO B.ByteString, IO ()) +provideActions wired_ptr = do + already_closed_mvar <- newMVar False + let + pushAction :: LB.ByteString -> IO () + pushAction datum = do + already_closed <- readMVar already_closed_mvar + if already_closed + then do + throwIO $ TLSLayerGenericProblem "Tried to send data on closed handle" + else do + BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + result <- sendData wired_ptr pchar (fromIntegral len) + case result of + r | r == allOk -> do + return () + | r == badHappened -> do + throwIO $ TLSLayerGenericProblem "Could not send data" + + pullAction :: IO B.ByteString + pullAction = do + already_closed <- readMVar already_closed_mvar + if already_closed + then do + throwIO $ TLSLayerGenericProblem "Tried to receive on closed handle" + else + allocaBytes useBufferSize $ \ pcharbuffer -> + alloca $ \ data_recvd_ptr -> do + result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr + recvd_bytes <- case result of + r | r == allOk -> peek data_recvd_ptr + | r == badHappened -> do + throwIO $ TLSLayerGenericProblem "Could not receive data" + + B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes) + + closeAction :: IO () + -- Ensure that the socket and the struct are only closed once + closeAction = do + b <- readMVar already_closed_mvar + if not b + then do + modifyMVar_ already_closed_mvar (\ _ -> return True) + disposeWiredSession wired_ptr + -- debugM "OpenSSL" "dispose clalled" + else + return () + return (pushAction, pullAction, closeAction) diff --git a/hs-src/SecondTransfer/Sessions/Config.hs b/hs-src/SecondTransfer/Sessions/Config.hs --- a/hs-src/SecondTransfer/Sessions/Config.hs +++ b/hs-src/SecondTransfer/Sessions/Config.hs @@ -1,14 +1,24 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-} +{- | Configuration and settings for the server. All constructor names are + exported, but notice that they start with an underscore. + They also have an equivalent lens without the + underscore. Please prefer to use the lens interface. +-} module SecondTransfer.Sessions.Config( sessionId ,defaultSessionsConfig + ,defaultSessionsEnrichedHeaders ,sessionsCallbacks + ,sessionsEnrichedHeaders ,reportErrorCallback + ,addUsedProtocol ,SessionComponent(..) ,SessionCoordinates(..) ,SessionsCallbacks(..) + ,SessionsEnrichedHeaders(..) + -- ,UsedProtocol(..) ,SessionsConfig(..) ,ErrorCallback ) where @@ -16,7 +26,7 @@ -- import Control.Concurrent.MVar (MVar) import Control.Exception (SomeException) -import Control.Lens (Lens', makeLenses) +import Control.Lens (makeLenses) -- | Information used to identify a particular session. @@ -38,7 +48,6 @@ fmap (\ s' -> (SessionCoordinates s')) (f session_id) - -- | Components at an individual session. Used to report -- where in the session an error was produced. This interface is likely -- to change in the future, as we add more metadata to exceptions @@ -51,6 +60,10 @@ deriving Show +-- Which protocol a session is using... no need for this right now +-- data UsedProtocol = +-- HTTP11_UsP +-- |HTTP2_UsP -- | Used by this session engine to report an error at some component, in a particular -- session. @@ -66,20 +79,40 @@ makeLenses ''SessionsCallbacks +-- | This is a temporal interface, but an useful one nonetheless. +-- By setting some values here to True, second-transfer will add +-- some headers to inbound requests, and some headers to outbound +-- requests. +data SessionsEnrichedHeaders = SessionsEnrichedHeaders { + -- | Adds a second-transfer-eh--used-protocol header + -- to inbound requests. Default: False + _addUsedProtocol :: Bool + } + +makeLenses ''SessionsEnrichedHeaders + +-- | Don't insert any extra-headers by default. +defaultSessionsEnrichedHeaders :: SessionsEnrichedHeaders +defaultSessionsEnrichedHeaders = SessionsEnrichedHeaders { + _addUsedProtocol = False + } + + -- | Configuration information you can provide to the session maker. data SessionsConfig = SessionsConfig { -- | Session callbacks _sessionsCallbacks :: SessionsCallbacks + ,_sessionsEnrichedHeaders :: SessionsEnrichedHeaders } --- makeLenses ''SessionsConfig +makeLenses ''SessionsConfig --- | Lens to access sessionsCallbacks in the `SessionsConfig` object. -sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks -sessionsCallbacks f ( - SessionsConfig { - _sessionsCallbacks= s - }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s) +-- -- | Lens to access sessionsCallbacks in the `SessionsConfig` object. +-- sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks +-- sessionsCallbacks f ( +-- SessionsConfig { +-- _sessionsCallbacks= s +-- }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s) -- | Creates a default sessions context. Modify as needed using @@ -88,6 +121,7 @@ defaultSessionsConfig = SessionsConfig { _sessionsCallbacks = SessionsCallbacks { _reportErrorCallback = Nothing - } + }, + _sessionsEnrichedHeaders = defaultSessionsEnrichedHeaders } diff --git a/hs-src/SecondTransfer/Utils/DevNull.hs b/hs-src/SecondTransfer/Utils/DevNull.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Utils/DevNull.hs @@ -0,0 +1,24 @@ + +module SecondTransfer.Utils.DevNull( + dropIncomingData + ) where + + +import Control.Concurrent (forkIO) +import Data.Conduit + +import SecondTransfer.MainLoop.CoherentWorker + +-- TODO: Handling unnecessary data should be done in some other, less +-- harmfull way... need to think about that. + +-- | If you are not processing the potential POST input in a request, +-- use this consumer to drop the data to oblivion. Otherwise it will +-- remain in an internal queue until the client closes the +-- stream, and if the client doesn't want to do so.... +dropIncomingData :: Maybe InputDataStream -> IO () +dropIncomingData Nothing = return () +dropIncomingData (Just data_source) = do + forkIO $ + data_source $$ (awaitForever (\ _ -> return () ) ) + return () diff --git a/hs-src/SecondTransfer/Utils/HTTPHeaders.hs b/hs-src/SecondTransfer/Utils/HTTPHeaders.hs --- a/hs-src/SecondTransfer/Utils/HTTPHeaders.hs +++ b/hs-src/SecondTransfer/Utils/HTTPHeaders.hs @@ -10,7 +10,8 @@ -- | These transformations are simple enough that don't require -- going away from the list representation (see type `Headers`) lowercaseHeaders - ,headersAreValidHTTP2 + ,headersAreLowercase + ,headersAreLowercaseAtHeaderEditor ,fetchHeader -- * Transformations based on maps -- @@ -44,9 +45,11 @@ import Data.Word (Word8) import Data.Time.Format (formatTime, defaultTimeLocale) -import Data.Time.Clock (UTCTime,getCurrentTime) +import Data.Time.Clock (getCurrentTime) +#ifndef IMPLICIT_MONOID import Control.Applicative ((<$>)) +#endif import SecondTransfer.MainLoop.CoherentWorker (Headers) @@ -63,15 +66,22 @@ -- | Checks that headers are lowercase -headersAreValidHTTP2 :: Headers -> Bool -headersAreValidHTTP2 headers = - let - isOk a_header = not . T.any isUpper . decodeUtf8 . fst $ a_header - in +headersAreLowercase :: Headers -> Bool +headersAreLowercase headers = foldl - (\ prev e -> (flip (&&)) (isOk e) $! prev) + (\ prev (hn, _) -> (flip (&&)) (aTitleIsLowercase hn) $! prev) True headers + +headersAreLowercaseAtHeaderEditor :: HeaderEditor -> Bool +headersAreLowercaseAtHeaderEditor header_editor = + Ms.foldlWithKey' + (\ prev hn _ -> (flip (&&)) (aTitleIsLowercase . toFlatBs $ hn) $! prev) + True + (innerMap header_editor) + +aTitleIsLowercase :: B.ByteString -> Bool +aTitleIsLowercase a_title = not . T.any isUpper . decodeUtf8 $ a_title -- | Looks for a given header diff --git a/second-transfer.cabal b/second-transfer.cabal --- a/second-transfer.cabal +++ b/second-transfer.cabal @@ -7,7 +7,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version : 0.5.2.2 +version : 0.5.3.1 synopsis : Second Transfer HTTP/2 web server @@ -53,7 +53,7 @@ source-repository this type: git location: git@github.com:alcidesv/second-transfer.git - tag: 0.5.2.2 + tag: 0.5.3.1 library @@ -64,6 +64,7 @@ , SecondTransfer.Exception , SecondTransfer.Types , SecondTransfer.Utils.HTTPHeaders + , SecondTransfer.Utils.DevNull -- These are really internal modules, but are exposed -- here for the sake of the test suite. They are hidden -- from the documentation. @@ -91,6 +92,8 @@ build-tools: cpphs + default-extensions: CPP + if flag(debug) CPP-Options: -DENABLE_DEBUG if !os(windows) @@ -98,6 +101,9 @@ else CC-Options: "-DNDEBUG" + if impl(ghc >= 7.10) + CPP-Options: -DIMPLICIT_MONOID -DIMPLICIT_APPLICATIVE_FOLDABLE + -- LANGUAGE extensions used by modules in this package. -- other-extensions: @@ -137,10 +143,12 @@ -- cc-options: -fPIC -pthread -g -O0 if flag(debug) cc-options: -O0 -g3 + ld-options: -g3 -- cc-options: -g3 -O0 - -- ghc-options: -O2 + -- ghc-options: -O2 -cpp -pgmPcpphs -optP--cpp + ghc-options: -pgmPcpphs -optP--cpp extra-libraries: ssl crypto diff --git a/tests/tests-hs-src/compiling_ok.hs b/tests/tests-hs-src/compiling_ok.hs --- a/tests/tests-hs-src/compiling_ok.hs +++ b/tests/tests-hs-src/compiling_ok.hs @@ -6,6 +6,7 @@ , tlsServeWithALPNAndFinishOnRequest , http2Attendant , http11Attendant + , dropIncomingData , FinishRequest(..) ) import SecondTransfer.Sessions( @@ -25,13 +26,15 @@ helloWorldWorker :: CoherentWorker -helloWorldWorker request = return ( - [ - (":status", "200") - ], - [], -- No pushed streams - saysHello - ) +helloWorldWorker (_request_headers, _maybe_post_data) = do + dropIncomingData _maybe_post_data + return ( + [ + (":status", "200") + ], + [], -- No pushed streams + saysHello + ) -- For this program to work, it should be run from the top of