diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -41,27 +41,13 @@ Example ------- -There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`. +There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`, and a somewhat more complicated one at +`examples/attempt_bust`; that one shows how to do HTTP/2.0 push from the library. -Roadmap -------- -Done: - -- Version 0.1: Having something that can run. No unit-testing, nothing - fancy. - -- Version 0.2: Absolutely minimal amount of unit tests. - -- Version 0.3: More sensible logging. - -Pending: - -- Benchmarking. - -Internal --------- +Development +----------- -Uploading documentation (provided you have access to the package): +Uploading documentation (provided you have access to the package in Hackage): $ ./hackage-upload-docs.sh second-transfer 0.5.4.0 diff --git a/cbits/tlsinc.c b/cbits/tlsinc.c --- a/cbits/tlsinc.c +++ b/cbits/tlsinc.c @@ -26,6 +26,50 @@ #include #include + +// + + +// Some constants +#define ALL_OK 0 +#define BAD_HAPPENED 1 +#define TIMEOUT_REACHED 3 +// This is also a failed IO with SSL, but this one may be quite +// natural and we want to handle it differently +#define TRANSPORT_CLOSED 2 + + +#if OPENSSL_VERSION_NUMBER < 0x1000200fL +#define ALPN_DISABLED +#warning Due to low OpenSSL version, this build is completely broken. Check second-transfer.cabal to see how +#warning to indicate a recent OpenSSL version. + +int SSL_CTX_set_ecdh_auto(SSL *s, int onoff) +{ + fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n"); +} + +void SSL_CTX_set_alpn_select_cb(SSL_CTX *ctx, + int (*cb) (SSL *ssl, + const unsigned char **out, + unsigned char *outlen, + const unsigned char *in, + unsigned int inlen, + void *arg), + void *arg) +{ + fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n"); +} + + +void SSL_get0_alpn_selected(SSL*a , int*b, int*c) +{ + fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n"); +} + +#endif + + // Simple structure to keep track of the handle, and // of what needs to be freed later. typedef struct { @@ -49,12 +93,7 @@ // Call when you are done void close_connection(connection_t* conn); // Wait for the next one... -#define ALL_OK 0 -#define BAD_HAPPENED 1 -#define TIMEOUT_REACHED 3 -// This is also a failed IO with SSL, but this one may be quite -// natural and we want to handle it differently -#define TRANSPORT_CLOSED 2 + int wait_for_connection(connection_t* conn, int microseconds, wired_session_t** wired_session); int send_data(wired_session_t* ws, char* buffer, int buffer_size); int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd); diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,8 @@ +- 0.7.1.0 : + * A few important memory leaks got fixed + * Changed I/O interface to live in IOCallbacks + * Working now around congestion window. + - 0.6.1.0 : * Improved documentation. * Improved ALPN negotiation handling. diff --git a/hs-src/SecondTransfer.hs b/hs-src/SecondTransfer.hs --- a/hs-src/SecondTransfer.hs +++ b/hs-src/SecondTransfer.hs @@ -6,14 +6,14 @@ Stability : experimental Portability : POSIX -SecondTransfer is a HTTP/1.1 and HTTP/2 server session library, with an emphasis towards +SecondTransfer is a HTTP\/1.1 and HTTP\/2 server session library, with an emphasis towards experimentation (so far). -This library implements enough of the HTTP/2 to build -compliant HTTP/2 servers. It also implements enough of -HTTP/1.1 so you can actually use it to build polyglot web-servers. +This library implements enough of the HTTP\/2 to build +compliant HTTP\/2 servers. It also implements enough of +HTTP\/1.1 so you can actually use it to build polyglot web-servers. -For HTTP/2, frame encoding and decoding is done with +For HTTP\/2, frame encoding and decoding is done with Kazu Yamamoto's package. This library just takes care of making sense of sent and received frames. @@ -111,7 +111,8 @@ most times you can do with a simplified version called `CoherentWorker`. The function `coherentToAwareWorker` does the conversion. The difference between the two callbacks is the level of information that you manage. With AwareWorker, you get a record on the request -with all sort of details, things like the session id and the +with all sort of details, things like the session id, the protocol the client is using and in +the future things like the remote address. The callback is used to handle all requests to the server on a given negotiated ALPN protocol. If you need routing functionality (and you most certainly will need it), you need 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 @@ -9,7 +9,9 @@ ,BadPrefaceException (..) ,HTTP11Exception (..) ,HTTP11SyntaxException (..) + ,ClientSessionAbortedException(..) ,HTTP500PrecursorException (..) + ,ConnectionCloseReason(..) ,convertHTTP500PrecursorExceptionToException ,getHTTP500PrecursorExceptionFromException ,ContentLengthMissingException (..) @@ -55,6 +57,25 @@ toException = convertHTTP2SessionExceptionToException fromException = getHTTP2SessionExceptionFromException +-- TODO below: add the other protocol reasons + +-- | Reasons for a remote server interrupting a connectionn of this client +data ConnectionCloseReason = + NormalTermination_CCR -- ^ Corresponds to NO_ERROR + |SessionAlreadyClosed_CCR -- ^ A request was done after the session was previously closed. + |IOChannelClosed_CCR -- ^ This one happens when one of the IO channels is closed and a BlockedIndefinitelyOnMVar bubbles up. It should only happen in the test suite, as the OpenSSL_TLS channel uses a specialized exception type. If you see it in the wild, it is a bug. + |ProtocolError_CCR -- ^ Any other reason + deriving Show + +-- | Concrete Exception. Used internally to signal that the server broke +-- the connection. This is a public exception that clients of the library +-- will see when acting as an HTTP client. +data ClientSessionAbortedException = ClientSessionAbortedException ConnectionCloseReason + deriving (Typeable, Show) + +instance Exception ClientSessionAbortedException where + toException = convertHTTP2SessionExceptionToException + fromException = getHTTP2SessionExceptionFromException -- | Abstract exception. Thrown when encoding/decoding of a frame fails data FramerException = forall e . Exception e => FramerException e diff --git a/hs-src/SecondTransfer/Http1/Session.hs b/hs-src/SecondTransfer/Http1/Session.hs --- a/hs-src/SecondTransfer/Http1/Session.hs +++ b/hs-src/SecondTransfer/Http1/Session.hs @@ -48,10 +48,10 @@ forkIO $ go new_session_tag (Just "") 1 return () where - push_action = attendant_callbacks ^. pushAction_AtC - -- pull_action = attendant_callbacks ^. pullAction_AtC - close_action = attendant_callbacks ^. closeAction_AtC - best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC + push_action = attendant_callbacks ^. pushAction_IOC + -- pull_action = attendant_callbacks ^. pullAction_IOC + close_action = attendant_callbacks ^. closeAction_IOC + best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_IOC go :: Int -> Maybe B.ByteString -> Int -> IO () go session_tag (Just leftovers) reuse_no = do diff --git a/hs-src/SecondTransfer/Http2/Framer.hs b/hs-src/SecondTransfer/Http2/Framer.hs --- a/hs-src/SecondTransfer/Http2/Framer.hs +++ b/hs-src/SecondTransfer/Http2/Framer.hs @@ -1,3 +1,4 @@ + -- 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, @@ -5,6 +6,7 @@ {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http2.Framer ( BadPrefaceException, + SessionPayload(..), wrapSession, http2FrameLength, @@ -46,6 +48,7 @@ import qualified Data.HashTable.IO as H import System.Clock (Clock(..),getTime) +-- import System.Mem.Weak import SecondTransfer.Sessions.Internal ( sessionExceptionHandler, @@ -119,19 +122,34 @@ data FramerSessionData = FramerSessionData { - _stream2flow :: MVar Stream2AvailSpace + + -- A dictionary (protected by a lock) from stream id to flow control command. + _stream2flow :: MVar Stream2AvailSpace + -- Two members below: the place where you put data which is going to be flow-controlled, -- and the place with the ordinal + -- Flow control dictionary. It goes from stream id to the stream data-output gate (an-mvar) + -- and to a mutable register for the ordinal. The ordinal for packets inside a stream is used + -- for priority and reporting , _stream2outputBytes :: MVar ( HashTable GlobalStreamId (MVar LB.ByteString, MVar Int) ) + + -- The default flow-control window advertised by the peer (e.g., the browser) , _defaultStreamWindow :: MVar Int - -- Wait variable to output bytes to the channel + -- Wait variable to output bytes to the channel. This is needed to avoid output data races. , _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 + + -- Exclusive lock for headers, needed since the specs forbid interleaving header and data , _noHeadersInChannel :: MVar NoHeadersInChannel + + -- The push action , _pushAction :: PushAction + + -- The close action , _closeAction :: CloseAction -- Global id of the session, used for e.g. error reporting. @@ -141,10 +159,19 @@ , _sessionsContext :: SessionsContext -- For GoAway frames - , _lastStream :: MVar Int + -- We keep the value below updated with the highest incoming stream + -- frame. For example, it is updated as soon as headers are received from + -- the client on that stream. + , _lastInputStream :: MVar Int + -- We update this one as soon as an outgoing frame is seen with such a + -- high output number. + , _lastOutputStream :: MVar Int -- For sending data orderly , _prioritySendState :: PrioritySendState + + -- Need to know this for the preface + , _sessionRole_FSD :: SessionRole } L.makeLenses ''FramerSessionData @@ -152,27 +179,41 @@ type FramerSession = ReaderT FramerSessionData IO +data SessionPayload = + AwareWorker_SP AwareWorker -- I'm a server + |ClientState_SP ClientState -- I'm a client -wrapSession :: AwareWorker -> SessionsContext -> Attendant -wrapSession aware_worker sessions_context attendant_callbacks = do +wrapSession :: SessionPayload -> SessionsContext -> Attendant +wrapSession session_payload sessions_context io_callbacks = do let session_id_mvar = view nextSessionId sessions_context - push_action = attendant_callbacks ^. pushAction_AtC - pull_action = attendant_callbacks ^. pullAction_AtC - close_action = attendant_callbacks ^. closeAction_AtC - best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC + push_action = io_callbacks ^. pushAction_IOC + pull_action = io_callbacks ^. pullAction_IOC + close_action = io_callbacks ^. closeAction_IOC + -- best_effort_pull_action = io_callbacks ^. bestEffortPullAction_IOC new_session_id <- modifyMVarMasked session_id_mvar $ \ session_id -> return $ session_id `seq` (session_id + 1, session_id) - (session_input, session_output) <- http2Session - aware_worker - new_session_id - sessions_context + (session_input, session_output) <- case session_payload of + AwareWorker_SP aware_worker -> http2ServerSession + aware_worker + new_session_id + sessions_context + ClientState_SP client_state -> http2ClientSession + client_state + new_session_id + sessions_context + let + session_role = case session_payload of + AwareWorker_SP _ -> Server_SR + ClientState_SP _ -> Client_SR + + -- TODO : Add type annotations.... s2f <- H.new stream2flow_mvar <- newMVar s2f @@ -182,6 +223,7 @@ can_output <- newMVar CanOutput no_headers_in_channel <- newMVar NoHeadersInChannel last_stream_id <- newMVar 0 + last_output_stream_id <- newMVar 0 output_is_forbidden <- newMVar False prio_mvar <- STM.atomically $ newTMVar PQ.empty @@ -200,12 +242,14 @@ ,_closeAction = close_action ,_sessionIdAtFramer = new_session_id ,_sessionsContext = sessions_context - ,_lastStream = last_stream_id + ,_lastInputStream = last_stream_id + ,_lastOutputStream = last_output_stream_id ,_outputIsForbidden = output_is_forbidden ,_prioritySendState = PrioritySendState { _semToSend = sem_to_send, _prioQ = prio_mvar } + ,_sessionRole_FSD = session_role } @@ -226,7 +270,7 @@ exc_handler :: Int -> SessionsContext -> FramerException -> IO () exc_handler x y e = do modifyMVar_ output_is_forbidden (\ _ -> return True) - INSTRUMENTATION( errorM "HTTP2.Framer" "Exception went up" ) + -- INSTRUMENTATION( errorM "HTTP2.Framer" "Exception went up" ) sessionExceptionHandler Framer_HTTP2SessionComponent x y e io_exc_handler :: Int -> SessionsContext -> IOProblem -> IO () @@ -265,6 +309,10 @@ GlobalStreamId -> Int -> FramerSession Bool +addCapacity _ 0 = + -- By the specs, a WINDOW_UPDATE with 0 of credit should be considered a protocol + -- error + return False addCapacity 0 delta_cap = -- TODO: Implement session flow control return True @@ -273,7 +321,7 @@ table_mvar <- view stream2flow val <- liftIO $ withMVar table_mvar $ \ table -> H.lookup table stream_id - last_stream_mvar <- view lastStream + last_stream_mvar <- view lastInputStream last_stream <- liftIO . readMVar $ last_stream_mvar case val of Nothing | stream_id > last_stream -> @@ -340,25 +388,30 @@ -- -- This function also does part of the flow control: it registers WindowUpdate frames and triggers -- quota updates on the streams. +-- +-- Also, when this function exits its caller will issue a close_action inputGatherer :: PullAction -> SessionInput -> FramerSession () inputGatherer pull_action session_input = do - -- We can start by reading off the prefix.... - prefix <- liftIO $ pull_action http2PrefixLength - if prefix /= NH2.connectionPreface - then do - sendGoAwayFrame NH2.ProtocolError - liftIO $ - -- We just use the GoAway frame, although this is awfully early - -- and probably wrong - throwIO BadPrefaceException - else - INSTRUMENTATION( debugM "HTTP2.Framer" "Prologue validated" ) + + session_role <- view sessionRole_FSD + + when (session_role == Server_SR) $ do + -- We can start by reading off the prefix.... + prefix <- liftIO $ pull_action http2PrefixLength + when (prefix /= NH2.connectionPreface) $ do + sendGoAwayFrame NH2.ProtocolError + liftIO $ + -- We just use the GoAway frame, although this is awfully early + -- and probably wrong + throwIO BadPrefaceException + let source::Source FramerSession (Maybe NH2.Frame) source = transPipe liftIO $ readNextFrame pull_action source $$ consume True where + sendToSession :: Bool -> InputFrame -> IO () sendToSession starting frame = -- print(NH2.streamId $ NH2.frameHeader frame) @@ -383,9 +436,15 @@ consume starting = do maybe_maybe_frame <- await - case maybe_maybe_frame of + output_is_forbidden_mvar <- view outputIsForbidden + output_is_forbidden <- liftIO $ readMVar output_is_forbidden_mvar + -- Consumption ends automatically when the output is forbidden, this might help avoiding + -- attacks where a peer refuses to close its socket. + unless output_is_forbidden $ case maybe_maybe_frame of + Just Nothing -> + -- Only way to get here is by a closed connection condition I guess. abortSession Just (Just right_frame) -> do @@ -395,7 +454,10 @@ -- Bookkeep the increase on bytes on that stream -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id) succeeded <- lift $ addCapacity stream_id (fromIntegral credit) - unless succeeded abortSession + if not succeeded then + abortSession + else + consume_continue frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do @@ -426,18 +488,18 @@ -- And send the frame down to the session, so that session specific settings -- can be applied. liftIO $ sendToSession starting $! frame - + consume_continue a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do -- Update the keep of last stream -- lift . startStreamOutputQueueIfNotExists (NH2.fromStreamIdentifier stream_id) priority - lift . updateLastStream $ stream_id + lift . updateLastInputStream $ stream_id -- Send frame to the session liftIO $ sendToSession starting a_frame - -- tail recursion: go again... - consume_continue + -- tail recursion: go again... + consume_continue Nothing -> -- We may as well exit this thread @@ -447,6 +509,13 @@ -- All the output frames come this way first outputGatherer :: SessionOutput -> FramerSession () outputGatherer session_output = do + + session_role <- view sessionRole_FSD + + -- When acting as a client, the first step is to send the prefix... + when (session_role == Client_SR) $ + pushPrefix + frame_sent_report_callback <- view $ sessionsContext . sessionsConfig . @@ -454,7 +523,6 @@ dataDeliveryCallback_SC session_id <- view sessionIdAtFramer - let dataForFrame p1 p2 = @@ -467,16 +535,24 @@ command_or_frame <- liftIO $ getFrameFromSession session_output case command_or_frame of - Left CancelSession_SOC -> do - -- The session wants to cancel things - INSTRUMENTATION( debugM "HTTP2.Framer" "CancelSession_SOC processed") + Left (CancelSession_SOC error_code) -> do + -- The session wants to cancel things as harshly as possible, send a GoAway frame with + -- the information I have here. + sendGoAwayFrame error_code releaseFramer + -- And this causes this thread to finish, so that no new frames + -- are taken from the session. Correspondingly, an exception is raised in + -- the session if it tries to write another frame - Left (FinishStream_SOC stream_id ) -> + Left (SpecificTerminate_SOC last_stream_id) -> do + -- This is used when the session wants to finish in a specific way. + sendSpecificTerminateGoAway last_stream_id + releaseFramer + + Left (FinishStream_SOC stream_id ) -> do -- Session knows that we are done with the given stream, and that we can release -- the flow control structures - -- NOTICE: Unfortunately, this doesn't work, so ignore the message for now - -- finishFlowControlForStream stream_id. + finishFlowControlForStream stream_id cont Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _), ef ) -> do @@ -546,11 +622,18 @@ loopPart session_id frame_sent_report_callback -updateLastStream :: GlobalStreamId -> FramerSession () -updateLastStream stream_id = do - last_stream_id_mvar <- view lastStream +updateLastInputStream :: GlobalStreamId -> FramerSession () +updateLastInputStream stream_id = do + last_stream_id_mvar <- view lastInputStream liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) + +updateLastOutputStream :: GlobalStreamId -> FramerSession () +updateLastOutputStream stream_id = do + last_stream_id_mvar <- view lastOutputStream + liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) + + startStreamOutputQueueIfNotExists :: GlobalStreamId -> Int -> FramerSession () startStreamOutputQueueIfNotExists stream_id priority = do table_mvar <- view stream2flow @@ -574,13 +657,12 @@ s2o_mvar <- view stream2outputBytes - liftIO . withMVar s2o_mvar $ \ s2o -> H.insert s2o stream_id (bytes_chan, ordinal_num) + liftIO . withMVar s2o_mvar $ {-# SCC e1 #-} \ s2o -> H.insert s2o stream_id (bytes_chan, ordinal_num) stream2flow_mvar <- view stream2flow - liftIO . withMVar stream2flow_mvar $ \ s2c -> H.insert s2c stream_id command_chan - + liftIO . withMVar stream2flow_mvar $ {-# SCC e2 #-} \ s2c -> H.insert s2c stream_id command_chan -- initial_cap_mvar <- view defaultStreamWindow initial_cap <- liftIO $ readMVar initial_cap_mvar @@ -592,23 +674,22 @@ -- 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 + close_action read_state <- ask liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT - (flowControlOutput stream_id priority initial_cap 0 "" command_chan bytes_chan) + ({-# SCC forkFlowControlOutput #-} flowControlOutput stream_id priority initial_cap 0 "" command_chan bytes_chan) read_state return (bytes_chan , command_chan) @@ -670,15 +751,31 @@ sendBytes bs +pushPrefix :: FramerSession () +pushPrefix = do + let bs = LB.fromStrict NH2.connectionPreface + sendBytes bs + + +-- Default sendGoAwayFrame. This one assumes that actions are taken as soon as stream is +-- received. This is a good default strategy for streams that cause things to happen, that +-- is, the ones with POST and GET. sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession () sendGoAwayFrame error_code = do - last_stream_id_mvar <- view lastStream + last_stream_id_mvar <- view lastInputStream last_stream_id <- liftIO $ readMVar last_stream_id_mvar pushFrame (NH2.EncodeInfo NH2.defaultFlags 0 Nothing) (NH2.GoAwayFrame last_stream_id error_code "") +sendSpecificTerminateGoAway :: GlobalStreamId -> FramerSession () +sendSpecificTerminateGoAway last_stream = + pushFrame + (NH2.EncodeInfo NH2.defaultFlags 0 Nothing) + (NH2.GoAwayFrame last_stream NH2.NoError "") + + -- From this point on data is really serialized, sendBytes :: LB.ByteString -> FramerSession () sendBytes bs = do @@ -699,14 +796,21 @@ -- -- There is one of these for each stream -- -flowControlOutput :: Int -> Int -> Int -> Int -> LB.ByteString -> MVar FlowControlCommand -> MVar LB.ByteString -> FramerSession () +flowControlOutput :: Int + -> Int + -> Int + -> Int + -> LB.ByteString + -> MVar FlowControlCommand + -> MVar LB.ByteString + -> FramerSession () flowControlOutput stream_id priority capacity ordinal leftovers commands_chan bytes_chan = - if leftovers == "" - then do + ordinal `seq` if leftovers == "" + then {-# SCC fcOBranch1 #-} do -- Get more data (possibly block waiting for it) - bytes_to_send <- liftIO $ takeMVar bytes_chan + bytes_to_send <- liftIO $ {-# SCC perfectlyHarmlessExceptionPoint #-} takeMVar bytes_chan flowControlOutput stream_id priority capacity ordinal bytes_to_send commands_chan bytes_chan - else do + else {-# SCC fcOBranch2 #-} do -- Length? let amount = fromIntegral (LB.length leftovers - 9) if amount <= capacity @@ -715,11 +819,11 @@ -- I can send ... if no headers are in process.... -- liftIO . logit $ "set-priority (stream_id, prio, ordinal) " `mappend` (pack . show) (__stream_id, priority, ordinal) withPrioritySend priority stream_id ordinal leftovers - flowControlOutput stream_id priority (capacity - amount) (ordinal+1) "" commands_chan bytes_chan + flowControlOutput stream_id priority (capacity - amount) (ordinal+1 ) "" commands_chan bytes_chan else do -- I can not send because flow-control is full, wait for a command instead - command <- liftIO $ takeMVar commands_chan - case command of + command <- liftIO $ {-# SCC t2 #-} takeMVar commands_chan + case {-# SCC t3 #-} command of AddBytes_FCM delta_cap -> -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id) flowControlOutput stream_id priority (capacity + delta_cap) ordinal leftovers commands_chan bytes_chan @@ -728,7 +832,8 @@ releaseFramer :: FramerSession () releaseFramer = -- Release any resources pending... - + -- This is not needed as of right now, since garbage collection works well. + -- I'm leaving it here just in case we need to re-activate it in the future. return () diff --git a/hs-src/SecondTransfer/Http2/MakeAttendant.hs b/hs-src/SecondTransfer/Http2/MakeAttendant.hs --- a/hs-src/SecondTransfer/Http2/MakeAttendant.hs +++ b/hs-src/SecondTransfer/Http2/MakeAttendant.hs @@ -4,24 +4,20 @@ ) where -import SecondTransfer.Http2.Framer (wrapSession) +import SecondTransfer.Http2.Framer (wrapSession, SessionPayload(..)) import SecondTransfer.Sessions.Internal (SessionsContext) import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.MainLoop.PushPullType (Attendant) --- | The type of this function is equivalent to: --- --- @ --- http2Attendant :: CoherentWorker -> AttendantCallbacks -> IO () --- @ +-- | -- --- Given a `CoherentWorker`, this function wraps it with flow control, multiplexing, +-- Given an `AwareWorker`, this function wraps it with flow control, multiplexing, -- and state maintenance needed to run an HTTP/2 session. -- --- Notice that this function is using HTTP/2 over TLS. We haven't implemented yet --- a session handling mechanism for HTTP/1.1 . +-- Notice that this function is using HTTP/2 over TLS. There is an equivalent for HTTP +-- 1.1. http2Attendant :: SessionsContext -> AwareWorker -> Attendant http2Attendant sessions_context coherent_worker attendant_callbacks = do let - attendant = wrapSession coherent_worker sessions_context + attendant = wrapSession (AwareWorker_SP coherent_worker) sessions_context attendant attendant_callbacks diff --git a/hs-src/SecondTransfer/Http2/Session.hs b/hs-src/SecondTransfer/Http2/Session.hs --- a/hs-src/SecondTransfer/Http2/Session.hs +++ b/hs-src/SecondTransfer/Http2/Session.hs @@ -4,11 +4,14 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-} {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http2.Session( - http2Session + http2ServerSession + ,http2ClientSession ,getFrameFromSession ,sendFirstFrameToSession ,sendMiddleFrameToSession ,sendCommandToSession + ,makeClientState + ,pendingRequests_ClS ,CoherentSession ,SessionInput(..) @@ -20,6 +23,8 @@ ,SessionsCallbacks(..) ,SessionsConfig(..) ,ErrorCallback + ,ClientState(..) + ,SessionRole(..) -- Internal stuff ,OutputFrame @@ -29,7 +34,7 @@ #include "Logging.cpphs" -- System grade utilities -import Control.Concurrent (ThreadId, forkIO, threadDelay) +import Control.Concurrent (ThreadId, forkIO) import Control.Concurrent.Chan import Control.Exception (throwTo) import qualified Control.Exception as E @@ -48,6 +53,7 @@ import qualified Data.HashTable.IO as H import qualified Data.IntSet as NS import Data.Maybe (isJust) +import Data.Typeable #ifndef IMPLICIT_MONOID import Data.Monoid (mappend) #endif @@ -67,6 +73,8 @@ import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.MainLoop.Tokens import SecondTransfer.MainLoop.Protocol +import SecondTransfer.MainLoop.ClientPetitioner + import SecondTransfer.Sessions.Config import SecondTransfer.Sessions.Internal (sessionExceptionHandler, SessionsContext, @@ -74,8 +82,10 @@ import SecondTransfer.Utils (unfoldChannelAndSource) import SecondTransfer.Exception import qualified SecondTransfer.Utils.HTTPHeaders as He -import SecondTransfer.MainLoop.Logging (logWithExclusivity, logit) +import SecondTransfer.MainLoop.Logging (logit) +--import Debug.Trace + -- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to -- use :-( type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload, Effect) @@ -93,11 +103,16 @@ -- end of data. Middle value is delay in microseconds type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString, Effect) - +-- What to do regarding headers data HeaderOutputMessage = + -- Send the headers of the principal stream NormalResponse_HM (GlobalStreamId, MVar HeadersSent, Headers, Effect) - -- + -- Send a push-promise |PushPromise_HM (GlobalStreamId, GlobalStreamId, Headers, Effect) + -- Send a reset stream notification for the stream given below + |ResetStream_HM (GlobalStreamId, Effect) + -- Send a GoAway, where last-stream is the stream given below. + |GoAway_HM (GlobalStreamId, Effect) -- Settings imposed by the peer @@ -179,13 +194,27 @@ deriving Show --- temporary +-- Session output commands indicating something to the Framer data SessionOutputCommand = - CancelSession_SOC + -- Command sent to the framer to close the session as harshly as possible. The framer + -- will pick its own last valid stream. + CancelSession_SOC NH2.ErrorCodeId + -- Command sent to the framer to close the session specifying a specific last-stream. + -- The reason used will be "NoError" + |SpecificTerminate_SOC GlobalStreamId + -- Command sent to the framer to notify it of a normal end of stream + -- condition, this is used to naturally close each stream. |FinishStream_SOC GlobalStreamId deriving Show +-- The role of a session is either server or client. There are small +-- differences between both. +data SessionRole = + Client_SR + |Server_SR + deriving (Eq,Show) + -- Here is how we make a session type SessionMaker = SessionsContext -> IO Session @@ -195,7 +224,78 @@ data PostInputMechanism = PostInputMechanism (MVar (Maybe B.ByteString), InputDataStream) +------------- Regarding client state +type Message = (Headers,InputDataStream) +type RequestResult = Either ConnectionCloseReason Message + +data ClientState = ClientState { + -- Holds a queue. The client here puts the message (the request) and an VAR + -- where it will receive the response. + _pendingRequests_ClS :: MVar (Message, MVar RequestResult) + + -- This is the id of the next available stream + ,_nextStream_ClS :: MVar Int + + -- Says if the client has been closed + ,_clientIsClosed_ClS :: MVar Bool + + -- A dictionary from stream id to the client which is waiting for the + -- message + ,_response2Waiter_ClS ::HashTable GlobalStreamId (MVar RequestResult) + } + +makeLenses ''ClientState + + +makeClientState :: IO ClientState +makeClientState = do + request_chan <- newEmptyMVar + next_stream_mvar <- newMVar 3 + client_is_closed_mvar <- newMVar False + new_h <- H.new + + return ClientState { + _pendingRequests_ClS = request_chan + ,_nextStream_ClS = next_stream_mvar + ,_response2Waiter_ClS = new_h + ,_clientIsClosed_ClS = client_is_closed_mvar + } + + +handleRequest' :: ClientState -> Headers -> InputDataStream -> IO (Headers,InputDataStream) +handleRequest' client_state headers input_data = runReaderT (handleRequest headers input_data) client_state + +type ClientMonad = ReaderT ClientState IO + +handleRequest :: Headers -> InputDataStream -> ClientMonad Message +handleRequest headers input_data = do + pending_requests <- view pendingRequests_ClS + response_mvar <- liftIO $ newEmptyMVar + + liftIO $ E.catch + (do + {-# SCC cause1 #-} putMVar pending_requests ((headers,input_data),response_mvar) + either_reason_or_message <- {-# SCC cause2 #-} liftIO $ takeMVar response_mvar + case either_reason_or_message of + Left break_reason -> E.throw $ ClientSessionAbortedException break_reason + Right message -> return message + ) + ( ( \ _ -> E.throw $ ClientSessionAbortedException SessionAlreadyClosed_CCR ):: E.BlockedIndefinitelyOnMVar -> IO Message ) + + + + +instance ClientPetitioner ClientState where + + request = handleRequest' + +-------------- Regarding client state + + +-- SessionData is the actual state of the session, including the channels to the framer +-- outside. +-- -- NH2.Frame != Frame data SessionData = SessionData { -- ATTENTION: Ignore the warning coming from here for now @@ -209,12 +309,15 @@ -- 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 @@ -229,8 +332,13 @@ -- I make copies of it in different contexts, and as needed. ,_forWorkerThread :: WorkerThreadEnvironment + -- When acting as a server, this is the handler for processing requests ,_awareWorker :: AwareWorker + -- When acting as a client, this is where new requests are taken + -- from + ,_simpleClient :: ClientState + -- Some streams may be cancelled ,_streamsCancelled :: MVar NS.IntSet @@ -251,15 +359,28 @@ -- What is the next stream available for push? ,_nextPushStream :: MVar Int + + -- What role does this session has? + ,_sessionRole :: SessionRole + } makeLenses ''SessionData +http2ServerSession :: AwareWorker -> Int -> SessionsContext -> IO Session +http2ServerSession a i sctx = http2Session Server_SR a (error "NotAClient") i sctx + + +http2ClientSession :: ClientState -> Int -> SessionsContext -> IO Session +http2ClientSession client_state session_id sctx = + http2Session Client_SR (error "NotAServer") client_state session_id sctx + + -- v- {headers table size comes here!!} -http2Session :: AwareWorker -> Int -> SessionsContext -> IO Session -http2Session aware_worker session_id sessions_context = do +http2Session :: SessionRole -> AwareWorker -> ClientState -> Int -> SessionsContext -> IO Session +http2Session session_role aware_worker client_state session_id sessions_context = do session_input <- newChan session_output <- newChan session_output_mvar <- newMVar session_output @@ -309,6 +430,7 @@ ,_stream2HeaderBlockFragment = stream_request_headers ,_forWorkerThread = for_worker_thread ,_awareWorker = aware_worker + ,_simpleClient = client_state ,_streamsCancelled = cancelled_streams_mvar ,_stream2PostInputMechanism = stream2postinputmechanism ,_stream2WorkerThread = stream2workerthread @@ -317,18 +439,60 @@ ,_sessionSettings = session_settings ,_lastGoodStream = last_good_stream_mvar ,_nextPushStream = next_push_stream + ,_sessionRole = session_role } let + io_exc_handler :: SessionComponent -> E.BlockedIndefinitelyOnMVar -> IO () + io_exc_handler _component _e = do + case session_role of + Server_SR -> do + -- sessionExceptionHandler component session_id sessions_context e + -- Just ignore this kind of exceptions here, very explicitly, as they should naturally + -- happen when the framer is closed + return () + + Client_SR -> do + clientSideTerminate client_state IOChannelClosed_CCR + + exc_handler :: SessionComponent -> HTTP2SessionException -> IO () - exc_handler component e = sessionExceptionHandler component session_id sessions_context e + exc_handler component e = do + case session_role of + Server_SR -> + sessionExceptionHandler component session_id sessions_context e + + Client_SR -> do + let + maybe_client_session_aborted :: Maybe ClientSessionAbortedException + maybe_client_session_aborted | HTTP2SessionException ee <- e = cast ee + + maybe_protocol_error :: Maybe HTTP2ProtocolException + maybe_protocol_error | HTTP2SessionException ee <- e = cast ee + + case maybe_client_session_aborted of + Just (ClientSessionAbortedException reason) -> do + clientSideTerminate client_state reason + + + Nothing -> + if isJust maybe_protocol_error + then + clientSideTerminate client_state ProtocolError_CCR + else do + E.throw 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 - ) + exc_guard component action = + E.catch + (E.catch + action + (\e -> do + -- INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" ) + exc_handler component e + ) + ) + (io_exc_handler component) use_chunk_length = sessions_context ^. sessionsConfig . dataFrameSize @@ -345,6 +509,13 @@ forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent $ dataOutputThread use_chunk_length data_output session_output_mvar + -- If I'm a client, I also need a thread to poll for requests + when (session_role == Client_SR) $ do + forkIO $ exc_guard SessionClientPollThread_HTTP2SessionComponent + $ sessionPollThread session_data headers_output + return () + + -- 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. @@ -362,18 +533,19 @@ -- these values in a closure... session_input <- view sessionInput - decode_headers_table_mvar <- view toDecodeHeaders - stream_request_headers <- view stream2HeaderBlockFragment + -- decode_headers_table_mvar <- view toDecodeHeaders + -- stream_request_headers <- view stream2HeaderBlockFragment cancelled_streams_mvar <- view streamsCancelled - coherent_worker <- view awareWorker + -- coherent_worker <- view awareWorker - for_worker_thread_uns <- view forWorkerThread + -- for_worker_thread_uns <- view forWorkerThread stream2workerthread <- view stream2WorkerThread - receiving_headers_mvar <- view receivingHeaders - last_good_stream_mvar <- view lastGoodStream - current_session_id <- view sessionIdAtSession + -- receiving_headers_mvar <- view receivingHeaders + -- last_good_stream_mvar <- view lastGoodStream + -- current_session_id <- view sessionIdAtSession - input <- liftIO $ readChan session_input + input <- {-# SCC session_input #-} liftIO $ readChan session_input + session_role <- view sessionRole case input of @@ -423,145 +595,19 @@ -- 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. - MiddleFrame_SIC 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 - -- 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( errorM "HTTP2.Session" "Protocol error: bad stream id") - closeConnectionBecauseIsInvalid NH2.ProtocolError - -- So, now we know, we are ignoring priority frames at the moment - -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame) - 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 ) - -- Lets get a time - headers_arrived_time <- liftIO $ getTime Monotonic - -- 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 -#if LOGIT_SWITCH_TIMINGS - let - (Just path) = He.fetchHeader header_list ":path" - liftIO $ logit $ (pack . show $ stream_id ) `mappend` " -> " `mappend` path -#endif - -- /DEBUG - -- 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 - - let - perception = Perception { - _startedTime_Pr = headers_arrived_time, - _streamId_Pr = stream_id, - _sessionId_Pr = current_session_id, - _protocol_Pr = Http2_HPV, - _anouncedProtocols_Pr = Nothing - } - request = Request { - _headers_RQ = header_list_after, - _inputData_RQ = post_data_source, - _perception_RQ = perception - } - - -- 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 - -- - -- NOTE: Some late internal errors from the worker thread are - -- handled here by clossing the session. - -- - -- TODO: Log exceptions handled here. - liftIO $ do - thread_id <- forkIO $ E.catch - (runReaderT - (workerThread - request - coherent_worker) - for_worker_thread - ) - ( - ( \ _ -> do - -- Actions to take when the thread breaks.... - writeChan session_input InternalAbort_SIC - ) - :: HTTP500PrecursorException -> IO () - ) - - H.insert stream2workerthread stream_id thread_id - - return () - else - -- Frame doesn't end the headers... it was added before... so - -- probably do nothing - return () + MiddleFrame_SIC frame | Just (_stream_id, _bytes) <- isAboutHeaders frame -> + case session_role of + Server_SR -> do + -- Just append the frames to streamRequestHeaders + serverProcessIncomingHeaders frame + continue - continue + Client_SR -> do + clientProcessIncomingHeaders frame + continue + -- Peer can order a stream aborted, meaning that we shall not send any more data + -- on it. MiddleFrame_SIC frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do let stream_id = streamIdFromFrame frame liftIO $ do @@ -659,6 +705,19 @@ handleSettingsFrame settings_list continue + MiddleFrame_SIC (NH2.Frame _ (NH2.GoAwayFrame _ _ _ )) + | Server_SR <- session_role -> do + -- I was sent a go away, so go-away... + closeConnectionBecauseIsInvalid NH2.NoError + return () + + | Client_SR <- session_role -> do + -- I was sent a go away, so go-away, but use the kind of exception + -- that will unwind the stack gracefully + closeConnectionForClient NH2.NoError + return () + + MiddleFrame_SIC somethingelse -> unlessReceivingHeaders $ do -- An undhandled case here.... -- liftIO . logit $ "Strange frame:" `mappend` (pack . show $ somethingelse) @@ -691,6 +750,257 @@ (NH2.SettingsFrame []) +serverProcessIncomingHeaders :: NH2.Frame -> ReaderT SessionData IO () +serverProcessIncomingHeaders frame | Just (stream_id, bytes) <- isAboutHeaders frame = do + + -- Just append the frames to streamRequestHeaders + opens_stream <- appendHeaderFragmentBlock stream_id bytes + receiving_headers_mvar <- view receivingHeaders + last_good_stream_mvar <- view lastGoodStream + for_worker_thread_uns <- view forWorkerThread + decode_headers_table_mvar <- view toDecodeHeaders + stream_request_headers <- view stream2HeaderBlockFragment + coherent_worker <- view awareWorker + current_session_id <- view sessionIdAtSession + session_input <- view sessionInput + stream2workerthread <- view stream2WorkerThread + + if opens_stream + then do + maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar + case maybe_rcv_headers_of of + Just _ -> do + -- Bad peer, it is already sending headers + -- and trying to open another one + {-# SCC ccB1 #-} 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( errorM "HTTP2.Session" "Protocol error: bad stream id") + {-# SCC ccB2 #-} closeConnectionBecauseIsInvalid NH2.ProtocolError + -- So, now we know, we are ignoring priority frames at the moment + -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame) + 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 ) + -- Lets get a time + headers_arrived_time <- liftIO $ getTime Monotonic + -- 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 +#if LOGIT_SWITCH_TIMINGS + let + (Just path) = He.fetchHeader header_list ":path" + liftIO $ logit $ (pack . show $ stream_id ) `mappend` " -> " `mappend` path +#endif + -- /DEBUG + -- 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 <- validateIncomingHeadersServer headers_editor + + good_headers <- case maybe_good_headers_editor of + Just yes_they_are_good -> return yes_they_are_good + Nothing -> {-# SCC ccB3 #-} 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 + + let + perception = Perception { + _startedTime_Pr = headers_arrived_time, + _streamId_Pr = stream_id, + _sessionId_Pr = current_session_id, + _protocol_Pr = Http2_HPV, + _anouncedProtocols_Pr = Nothing + } + request = Request { + _headers_RQ = header_list_after, + _inputData_RQ = post_data_source, + _perception_RQ = perception + } + + -- 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 + -- + -- NOTE: Some late internal errors from the worker thread are + -- handled here by clossing the session. + -- + -- TODO: Log exceptions handled here. + liftIO $ do + thread_id <- forkIO $ E.catch + (runReaderT + (workerThread + request + coherent_worker) + for_worker_thread + ) + ( + ( \ _ -> do + -- Actions to take when the thread breaks.... + writeChan session_input InternalAbort_SIC + ) + :: HTTP500PrecursorException -> IO () + ) + + H.insert stream2workerthread stream_id thread_id + + return () + else + -- Frame doesn't end the headers... it was added before... so + -- probably do nothing + return () + + +clientProcessIncomingHeaders :: NH2.Frame -> ReaderT SessionData IO () +clientProcessIncomingHeaders frame | Just (stream_id, bytes) <- isAboutHeaders frame = do + + opens_stream <- appendHeaderFragmentBlock stream_id bytes + receiving_headers_mvar <- view receivingHeaders + last_good_stream_mvar <- view lastGoodStream + decode_headers_table_mvar <- view toDecodeHeaders + stream_request_headers <- view stream2HeaderBlockFragment + response2waiter <- view (simpleClient . response2Waiter_ClS) + + if opens_stream + then do + maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar + case maybe_rcv_headers_of of + Just _ -> do + -- Bad peer, 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 + stream_initiated_by_client <- streamInitiatedByClient stream_id + if (odd stream_id ) && stream_initiated_by_client + then do + -- We are golden, set the new good stream + liftIO $ putMVar last_good_stream_mvar (stream_id) + else do + -- We are not golden + -- TODO: Control for pushed streams here. + closeConnectionBecauseIsInvalid NH2.ProtocolError + -- So, now we know, we are ignoring priority frames at the moment + -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame) + 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 ) + -- Lets get a time + -- headers_arrived_time <- liftIO $ getTime Monotonic + -- Let's decode the headers + 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 <- validateIncomingHeadersClient headers_editor + + good_headers <- case maybe_good_headers_editor of + Just yes_they_are_good -> return yes_they_are_good + Nothing -> closeConnectionBecauseIsInvalid NH2.ProtocolError + + + -- If the headers end the request.... + post_data_source <- if not (frameEndsStream frame) + then do + mechanism <- createMechanismForStream stream_id + return $ postDataSourceFromMechanism mechanism + else + return $ return () + + (Just response_mvar) <- liftIO $ H.lookup response2waiter stream_id + liftIO $ putMVar response_mvar $ Right (He.toList good_headers, post_data_source) + + return () + else + -- Frame doesn't end the headers... it was added before... so + -- probably do nothing + return () + + +streamInitiatedByClient :: GlobalStreamId -> ReaderT SessionData IO Bool +streamInitiatedByClient stream_id = do + response2waiter <- view (simpleClient . response2Waiter_ClS) + response_handle_maybe <- liftIO $ H.lookup response2waiter stream_id + return $ isJust response_handle_maybe + + sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO () sendOutFrame encode_info payload = do session_output_mvar <- view sessionOutput @@ -730,8 +1040,9 @@ else return headers_editor -validateIncomingHeaders :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor) -validateIncomingHeaders headers_editor = do +-- TODO: Close connection on unexepcted pseudo-headers +validateIncomingHeadersServer :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor) +validateIncomingHeadersServer 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... -- @@ -757,7 +1068,27 @@ else return Nothing +validateIncomingHeadersClient :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor) +validateIncomingHeadersClient 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_status = h1 ^. (He.headerLens ":status") + if (isJust maybe_status) + 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 @@ -791,7 +1122,7 @@ -- that it stops accepting frames from connected sources -- (Streams?) session_output <- takeMVar session_output_mvar - writeChan session_output $ Left CancelSession_SOC + writeChan session_output $ Left (CancelSession_SOC error_code) putMVar session_output_mvar session_output -- And unwind the input thread in the session, so that the @@ -799,6 +1130,59 @@ E.throw HTTP2ProtocolException +-- Sends a GO_AWAY frame and raises an exception, effectively terminating the input +-- thread of the session. This one for the client is different because it throws +-- an exception of type ClientSessionAbortedException +closeConnectionForClient :: NH2.ErrorCodeId -> ReaderT SessionData IO a +closeConnectionForClient error_code = do + let + use_reason = case error_code of + NH2.NoError -> NormalTermination_CCR + _ -> ProtocolError_CCR + + -- 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 + 0 + Nothing + ) + (NH2.GoAwayFrame + last_good_stream + error_code + "" + ) + + client_is_closed_mvar <- view (simpleClient . clientIsClosed_ClS ) + + 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 $ error_code + putMVar session_output_mvar session_output + + -- Let's also mark the session as closed from the client side, so that + -- any further requests end with the correct exception + modifyMVar_ client_is_closed_mvar (\ _ -> return True) + + -- And unwind the input thread in the session, so that the + -- exception handler runs.... + E.throw $ ClientSessionAbortedException use_reason + + frameEndsStream :: InputFrame -> Bool frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags @@ -839,7 +1223,10 @@ case pim_maybe of Just (PostInputMechanism (chan, _)) -> - liftIO $ putMVar chan Nothing + liftIO $ do + putMVar chan Nothing + -- Not sure if this will work + H.delete stream2postinputmechanism stream_id Nothing -> -- TODO: This is a protocol error, handle it properly @@ -929,6 +1316,37 @@ -- Pieces of the header let + effects = principal_stream ^. effect_PS + interrupt_maybe = effects ^. interrupt_Ef + + + + case interrupt_maybe of + + Nothing -> do + normallyHandleStream principal_stream + + Just (InterruptConnectionAfter_IEf) -> do + normallyHandleStream principal_stream + liftIO . writeChan headers_output $ GoAway_HM (stream_id, effects) + + Just (InterruptConnectionNow_IEf) -> do + -- Not one hundred-percent sure of this being correct, but we don't want + -- to acknowledge reception of this stream then + let use_stream_id = stream_id - 1 + liftIO . writeChan headers_output $ GoAway_HM (use_stream_id, effects) + + +normallyHandleStream :: PrincipalStream -> WorkerMonad () +normallyHandleStream principal_stream = do + headers_output <- view headersOutput + stream_id <- view streamId + session_settings_mvar <- view sessionSettings_WTE + session_settings <- liftIO $ readMVar session_settings_mvar + next_push_stream_mvar <- view nextPushStream_WTE + + -- Pieces of the header + let headers = principal_stream ^. headers_PS data_and_conclusion = principal_stream ^. dataAndConclusion_PS effects = principal_stream ^. effect_PS @@ -936,6 +1354,8 @@ can_push = session_settings ^. pushEnabled + -- This gets executed in normal conditions, when no interruption is required. + -- There are several possible moments where the PUSH_PROMISEs can be sent, -- but a default safe one is before sending the response HEADERS, so that -- LINK headers in the response come after any potential promises. @@ -1060,10 +1480,13 @@ Nothing -> -- This is how we finish sending data liftIO $ putMVar data_output (stream_id, Nothing, effect) - Just bytes -> do - liftIO $ do - putMVar data_output (stream_id, Just bytes, effect) - consumer data_output + Just bytes + | B.length bytes > 0 -> do + liftIO $ do + putMVar data_output (stream_id, Just bytes, effect) + consumer data_output + | otherwise -> + liftIO $ putMVar data_output (stream_id, Nothing, effect) -- Returns if the frame is the first in the stream @@ -1121,8 +1544,8 @@ use_chunk_length <- view $ sessionsContext . sessionsConfig . dataFrameSize - header_output_request <- liftIO $ readChan input_chan - case header_output_request of + header_output_request <- {-# SCC input_chan #-} liftIO $ readChan input_chan + {-# SCC case_ #-} case header_output_request of NormalResponse_HM (stream_id, headers_ready_mvar, headers, effect) -> do -- First encode the headers using the table @@ -1148,6 +1571,15 @@ putMVar headers_ready_mvar HeadersSent ) + GoAway_HM (stream_id, _effect) -> do + -- This is in charge of sending an interrupt message to the framer + let + message = Left (SpecificTerminate_SOC stream_id) + liftIO . withMVar session_output_mvar $ + (\session_output -> do + writeChan session_output message + ) + PushPromise_HM (parent_stream_id, child_stream_id, promise_headers, effect) -> do encode_dyn_table_mvar <- view toEncodeHeaders @@ -1287,14 +1719,111 @@ writeContinuations fragments stream_id effect = mapM_ (\ fragment -> withLockedSessionOutput - (\ session_output -> writeChan session_output $ Right ( - NH2.EncodeInfo { - NH2.encodeFlags = NH2.defaultFlags - ,NH2.encodeStreamId = stream_id - ,NH2.encodePadding = Nothing - }, - NH2.DataFrame fragment, - effect ) + (\ session_output -> do + -- TODO: + when (B.length fragment == use_chunk_length) $ do + -- Force tons and tons of ping frames to see if we get less delays due to + -- congestion... + writeChan session_output $ Right ( + NH2.EncodeInfo { + NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = 0 + ,NH2.encodePadding = Nothing + }, + NH2.PingFrame "talk on", + effect ) + + writeChan session_output $ Right ( + NH2.EncodeInfo { + NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = stream_id + ,NH2.encodePadding = Nothing + }, + NH2.DataFrame fragment, + effect ) ) ) fragments + + +sessionPollThread :: SessionData -> Chan HeaderOutputMessage -> IO () +sessionPollThread session_data headers_output = do + let + pending_requests_mvar = session_data ^. (simpleClient . pendingRequests_ClS) + client_next_stream_mvar = session_data ^. (simpleClient . nextStream_ClS) + response2waiter = session_data ^. (simpleClient . response2Waiter_ClS) + for_worker_thread = session_data ^. forWorkerThread + session_input = session_data ^. sessionInput + + ((headers, input_data_stream), response_mvar) <- takeMVar pending_requests_mvar + + new_stream_id <- modifyMVar client_next_stream_mvar (\ n -> return (n + 1, n) ) + let + worker_environment = set streamId new_stream_id for_worker_thread + effects = defaultEffects + + -- Store the response place + new_stream_id `seq` H.insert response2waiter new_stream_id response_mvar + + -- Start by sending the headers of the request + headers_sent <- liftIO newEmptyMVar + liftIO $ writeChan headers_output $ NormalResponse_HM (new_stream_id, headers_sent, headers, effects) + + liftIO . forkIO $ E.catch + (runReaderT + (clientWorkerThread new_stream_id effects headers_sent input_data_stream ) + worker_environment + ) + ( + ( \ _ -> do + -- Actions to take when the thread breaks.... + writeChan session_input InternalAbort_SIC + ) + :: HTTP500PrecursorException -> IO () + ) + + sessionPollThread session_data headers_output + + +clientWorkerThread :: GlobalStreamId -> Effect -> MVar HeadersSent -> InputDataStream -> WorkerMonad () +clientWorkerThread stream_id effects headers_sent input_data_stream = do + -- And now work on sending the data, if any... + runConduit $ + transPipe liftIO input_data_stream + `fuseBothMaybe` + sendDataOfStream stream_id headers_sent effects + return () + + +clientSideTerminate :: ClientState -> ConnectionCloseReason -> IO () +clientSideTerminate client_state reason = do + let + terminateOnQueue :: ConnectionCloseReason -> MVar (Message, MVar RequestResult) -> IO () + terminateOnQueue reason' q = do + w <- tryTakeMVar q + case w of + Just (_, m) -> do + tryPutMVar m (Left reason') + return () + + Nothing -> + return () + + + terminate :: ConnectionCloseReason -> IO () + terminate reason' = do + terminateOnQueue reason' (client_state ^. pendingRequests_ClS) + + -- Kill any waiters in the map + H.mapM_ + (\(_k,v) -> tryPutMVar v (Left reason)) + (client_state ^. response2Waiter_ClS) + + client_is_closed_mvar = client_state ^. clientIsClosed_ClS + + -- Let's also mark the session as closed from the client side, so that + -- any further requests end with the correct exception + modifyMVar_ client_is_closed_mvar $ \ client_is_closed -> do + unless client_is_closed $ + terminate reason + return True diff --git a/hs-src/SecondTransfer/Http2/SimpleClient.hs b/hs-src/SecondTransfer/Http2/SimpleClient.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/SimpleClient.hs @@ -0,0 +1,35 @@ +{-# LANGUAGE OverloadedStrings, TemplateHaskell #-} +-- TODO: Remove this file, it is not needed!! +module SecondTransfer.Http2.SimpleClient ( + ClientState(..) -- reduced functionality is exported to the user, but + -- the framework may need a lot. + ,pendingRequests_ClS + ) where + + +import Control.Lens +import Control.Concurrent.Chan +-- 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,unless, when, mapM_, forM, forM_) +import Control.Monad.IO.Class (liftIO) +import Control.DeepSeq ( ($!!), deepseq ) +import Control.Monad.Trans.Reader +-- import Control.Monad.Catch (throwM) +import Control.Concurrent.MVar + + +import Data.Conduit +--import + +import SecondTransfer.MainLoop.ClientPetitioner +import SecondTransfer.MainLoop.CoherentWorker (Headers,InputDataStream) +import SecondTransfer.Http2.Session (ClientState(..), + pendingRequests_ClS + ) + +void::() +void = () diff --git a/hs-src/SecondTransfer/MainLoop/ClientPetitioner.hs b/hs-src/SecondTransfer/MainLoop/ClientPetitioner.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/ClientPetitioner.hs @@ -0,0 +1,15 @@ +module SecondTransfer.MainLoop.ClientPetitioner( + ClientPetitioner(..) + ) where + + +--import Control.Lens +--import Data.Conduit +import qualified Data.ByteString as B + + +import SecondTransfer.MainLoop.CoherentWorker (Headers,InputDataStream) + + +class ClientPetitioner a where + request :: a -> Headers -> InputDataStream -> IO (Headers,InputDataStream) 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 @@ -29,6 +29,7 @@ , TupledPrincipalStream , TupledRequest , FragmentDeliveryCallback + , InterruptEffect(..) , headers_RQ , inputData_RQ @@ -47,6 +48,7 @@ , protocol_Pr , fragmentDeliveryCallback_Ef , priorityEffect_Ef + , interrupt_Ef , defaultEffects , coherentToAwareWorker @@ -83,7 +85,7 @@ type Headers = [Header] -- |This is a Source conduit (see Haskell Data.Conduit library from Michael Snoyman) --- that you can use to retrieve the data sent by the client piece-wise. +-- that you can use to retrieve the data sent by the peer piece-wise. type InputDataStream = Source IO B.ByteString @@ -157,16 +159,30 @@ type FragmentDeliveryCallback = Int -> TimeSpec -> IO () +-- | Types of interrupt effects that can be signaled by aware workers. These include whole +-- connection shutdowns and stream resets. In all the cases, the reason given will be +-- NO_ERROR. +data InterruptEffect = InterruptConnectionAfter_IEf -- ^ Close and send GoAway /after/ this stream finishes delivery + |InterruptConnectionNow_IEf -- ^ Close and send GoAway /without/ delivering this stream. This implies that + -- other fields of the PrincipalStream record will be ignored. + -- |InterruptThisStream_IEf -- ^ Just reset this stream, disabled for now. + + -- | Sometimes a response needs to be handled a bit specially, -- for example by reporting delivery details back to the worker data Effect = Effect { + -- | A callback to be called whenever a data-packet for this stream is called. _fragmentDeliveryCallback_Ef :: Maybe FragmentDeliveryCallback - -- In certain circunstances a stream can use an internal priority, + -- | In certain circunstances a stream can use an internal priority, -- not given by the browser and the protocol. Lowest values here are -- given more priority. Default (when Nothing) is given zero. Cases -- with negative numbers also work. ,_priorityEffect_Ef :: Maybe Int + + -- | There are situations when it is desirable to close a stream or the entire + -- connection. Use this member to indicate that. + ,_interrupt_Ef :: Maybe InterruptEffect } makeLenses ''Effect @@ -174,13 +190,14 @@ defaultEffects :: Effect defaultEffects = Effect { _fragmentDeliveryCallback_Ef = Nothing, - _priorityEffect_Ef = Nothing + _priorityEffect_Ef = Nothing, + _interrupt_Ef = Nothing } -- | You use this type to answer a request. The `Headers` are thus response -- headers and they should contain the :status pseudo-header. The `PushedStreams` --- is a list of pushed streams...(I don't thaink that I'm handling those yet) +-- is a list of pushed streams... they will be pushed to the client. data PrincipalStream = PrincipalStream { _headers_PS :: Headers, _pushedStreams_PS :: PushedStreams, diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs --- a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +++ b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs @@ -301,7 +301,7 @@ smallWaitTime = 50000 -- provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), Int -> IO B.ByteString, IO ()) -provideActions :: Wired_Ptr -> IO AttendantCallbacks +provideActions :: Wired_Ptr -> IO IOCallbacks provideActions wired_ptr = do can_write_mvar <- newMVar True can_read_mvar <- newMVar True @@ -370,9 +370,9 @@ else return (False,False) - return AttendantCallbacks { - _pushAction_AtC = pushAction, - _pullAction_AtC = pullAction, - _closeAction_AtC = closeAction, - _bestEffortPullAction_AtC = bestEffortPullAction + return IOCallbacks { + _pushAction_IOC = pushAction, + _pullAction_IOC = pullAction, + _closeAction_IOC = closeAction, + _bestEffortPullAction_IOC = bestEffortPullAction } diff --git a/hs-src/SecondTransfer/MainLoop/PushPullType.hs b/hs-src/SecondTransfer/MainLoop/PushPullType.hs --- a/hs-src/SecondTransfer/MainLoop/PushPullType.hs +++ b/hs-src/SecondTransfer/MainLoop/PushPullType.hs @@ -3,14 +3,15 @@ module SecondTransfer.MainLoop.PushPullType ( PushAction ,PullAction + ,BestEffortPullAction ,Attendant ,CloseAction - ,AttendantCallbacks(..) + ,IOCallbacks(..) - ,pushAction_AtC - ,pullAction_AtC - ,closeAction_AtC - ,bestEffortPullAction_AtC + ,pushAction_IOC + ,pullAction_IOC + ,closeAction_IOC + ,bestEffortPullAction_IOC ) where @@ -49,22 +50,22 @@ -- | A set of functions describing how to do I/O in a session. -- As usual, we provide lenses accessors. -data AttendantCallbacks = AttendantCallbacks { +data IOCallbacks = IOCallbacks { -- | put some data in the channel - _pushAction_AtC :: PushAction, + _pushAction_IOC :: PushAction, -- | get exactly this much data from the channel. This function can -- be used by HTTP/2 since lengths are pretty well built inside the -- protocoll itself. - _pullAction_AtC :: PullAction, + _pullAction_IOC :: PullAction, -- | pull data from the channel, as much as the TCP stack wants to provide. -- we have no option but use this one when talking HTTP/1.1, where the best -- way to know the length is to scan until a Content-Length is found. - _bestEffortPullAction_AtC :: BestEffortPullAction, + _bestEffortPullAction_IOC :: BestEffortPullAction, -- | this is called when we wish to close the channel. - _closeAction_AtC :: CloseAction + _closeAction_IOC :: CloseAction } -makeLenses ''AttendantCallbacks +makeLenses ''IOCallbacks -- | This is an intermediate type. It represents what you obtain -- by combining something that speaks the protocol and an AwareWorker. @@ -83,4 +84,4 @@ -- This library supplies two of such Attendant factories, -- 'SecondTransfer.Http1.http11Attendant' for -- HTTP 1.1 sessions, and 'SecondTransfer.Http2.http2Attendant' for HTTP/2 sessions. -type Attendant = AttendantCallbacks -> IO () +type Attendant = IOCallbacks -> IO () 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 @@ -59,6 +59,7 @@ SessionInputThread_HTTP2SessionComponent |SessionHeadersOutputThread_HTTP2SessionComponent |SessionDataOutputThread_HTTP2SessionComponent + |SessionClientPollThread_HTTP2SessionComponent |Framer_HTTP2SessionComponent |Session_HTTP11 deriving Show diff --git a/hs-src/SecondTransfer/Sessions/Internal.hs b/hs-src/SecondTransfer/Sessions/Internal.hs --- a/hs-src/SecondTransfer/Sessions/Internal.hs +++ b/hs-src/SecondTransfer/Sessions/Internal.hs @@ -10,7 +10,7 @@ import Control.Lens ((^.), makeLenses) -import System.Log.Logger +import System.Log.Logger @@ -45,7 +45,6 @@ } - sessionExceptionHandler :: E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO () sessionExceptionHandler session_component session_id sessions_context e = @@ -61,7 +60,9 @@ ) in case maybe_error_callback of Nothing -> - errorM component_tag (show e) + -- errorM component_tag (show e) + -- When no callback, ignore the exception + return () Just callback -> callback error_tuple 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.6.1.0 +version : 0.7.1.0 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.6.1.0 + tag: 0.7.1.0 library @@ -79,6 +79,8 @@ , SecondTransfer.MainLoop.Tokens , SecondTransfer.MainLoop.Framer , SecondTransfer.MainLoop.Logging + , SecondTransfer.MainLoop.ClientPetitioner + , SecondTransfer.Sessions.Internal , SecondTransfer.MainLoop.OpenSSL_TLS @@ -87,6 +89,7 @@ , SecondTransfer.Http2.Framer , SecondTransfer.Http2.MakeAttendant , SecondTransfer.Http2.Session + , SecondTransfer.Http2.SimpleClient , SecondTransfer.Http1.Session @@ -139,26 +142,25 @@ -- Base language which the package is written in. default-language: Haskell2010 - -- Very specific directory with the version of openssl - -- that I'm using. As of February 2015-- this one is not - -- commonly installed + -- NOTE: Very specific directory with the version of openssl + -- that I'm using. As of February 2015-- openssl 1.0.2 is not + -- commonly installed. Update this path in your build + -- or otherwise your build will be broken. include-dirs: /opt/openssl-1.0.2/include c-sources: cbits/tlsinc.c - -- cc-options: -fPIC -pthread -g -O0 if flag(debug) cc-options: -O0 -g3 ld-options: -g3 - -- cc-options: -g3 -O0 - - -- ghc-options: -O2 -cpp -pgmPcpphs -optP--cpp ghc-options: -pgmPcpphs -optP--cpp extra-libraries: ssl crypto - -- NOTICE: Please fill-in with an-up-to date library path here + -- NOTE: Please fill-in with an-up-to date library path here. + -- It should point to openssl 1.0.2 or greater. See note for + -- include-dirs above. extra-lib-dirs: /opt/openssl-1.0.2/lib include-dirs: macros/ @@ -177,6 +179,7 @@ + Test-Suite hunit-tests type : exitcode-stdio-1.0 main-is : hunit_tests.hs @@ -230,9 +233,14 @@ , SecondTransfer.MainLoop.Tokens , SecondTransfer.MainLoop.Framer , SecondTransfer.MainLoop.Logging + , SecondTransfer.MainLoop.Protocol , SecondTransfer.Sessions.Internal + , SecondTransfer.MainLoop.ClientPetitioner , SecondTransfer.Utils , SecondTransfer.Http2.Framer , SecondTransfer.Http2.MakeAttendant , SecondTransfer.Http2.Session , SecondTransfer.Http1.Session + , Tests.HTTP1Parse + , Tests.HTTP2Session + , Tests.Utils diff --git a/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs b/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs --- a/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs +++ b/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs @@ -5,79 +5,203 @@ import Control.Concurrent -import Control.Concurrent.MVar -import Control.Concurrent.Chan +import Control.Concurrent.STM.TMVar +import Control.Concurrent.STM.TChan import Control.Exception -import Control.Lens (view, (^.)) -import qualified Control.Lens as L +import Control.Lens (view, (^.), over) +import qualified Control.Lens as L +import Control.Monad.STM (atomically) -import qualified Network.HTTP2 as NH2 -import qualified Network.HPACK as HP +import qualified Network.HTTP2 as NH2 +import qualified Network.HPACK as HP -import qualified Data.ByteString as B -import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as LB +import qualified Data.ByteString.Builder as Bu +import qualified Data.Sequence as Sq +import Data.Sequence ( (<|), (|>), ViewL(..), Seq ) import SecondTransfer.Http2 import SecondTransfer.Types +import SecondTransfer.Http2.Session +import SecondTransfer.Sessions.Config +import SecondTransfer.Sessions.Internal +import SecondTransfer.Http2.Framer import SecondTransfer.MainLoop.Internal +import Debug.Trace +-- "Internal imports" +import SecondTransfer.MainLoop.CoherentWorker (defaultEffects) + +data DataContinuityEngine = DataContinuityEngine { + _incoming_DCE :: Seq B.ByteString + } + + +L.makeLenses ''DataContinuityEngine + + +emptyDCE :: DataContinuityEngine +emptyDCE = DataContinuityEngine Sq.empty + + +dceAddData :: DataContinuityEngine -> B.ByteString -> DataContinuityEngine +dceAddData dce moredata | B.length moredata > 0 = over incoming_DCE ( |> moredata ) dce +dceAddData dce moredata | otherwise = dce + + +tryGetThisManyData :: DataContinuityEngine -> Int -> Maybe (DataContinuityEngine, B.ByteString) +tryGetThisManyData dce n = let + go_result = go n "" (dce ^. incoming_DCE) + go :: Int -> Bu.Builder -> Seq B.ByteString -> Maybe (Seq B.ByteString, Bu.Builder) + go n bld seq | Sq.EmptyL <- Sq.viewl seq, n > 0 = Nothing + | Sq.EmptyL <- Sq.viewl seq, n == 0 = Just (seq, bld) + | lead :< rest <- Sq.viewl seq, n > B.length lead = go (n - B.length lead) (bld `mappend` Bu.byteString lead) rest + | lead :< rest <- Sq.viewl seq, n == B.length lead = Just ( rest, bld `mappend` Bu.byteString lead) + | lead :< rest <- Sq.viewl seq, n < B.length lead = let + (tk, lv) = B.splitAt n lead + in Just ( lv <| rest, bld `mappend` Bu.byteString tk ) + in case go_result of + Just (new_dce, bld ) -> Just + (DataContinuityEngine new_dce, LB.toStrict . Bu.toLazyByteString $ bld ) + + Nothing -> Nothing + + +dceIsEmpty :: DataContinuityEngine -> Bool +dceIsEmpty dce = Sq.null $ dce ^. incoming_DCE + + +dceGetAllData :: DataContinuityEngine -> (DataContinuityEngine, B.ByteString) +dceGetAllData dce = (emptyDCE, all_together) + where + all_together = LB.toStrict . Bu.toLazyByteString . L.foldMapOf (incoming_DCE . L.folded) Bu.byteString $ dce + + data DecoySession = DecoySession { - _inputDataChannel :: Chan B.ByteString - ,_outputDataChannel :: Chan LB.ByteString - ,_sessionThread :: ThreadId - ,_remainingOutputBit:: MVar B.ByteString - ,_nh2Settings :: NH2.Settings - ,_sessionThrowed :: MVar Bool - ,_waiting :: MVar ThreadId - ,_encodeHeadersHere :: MVar HP.DynamicTable - ,_decodeHeadersHere :: MVar HP.DynamicTable + _inputDataChannel :: TChan B.ByteString + ,_outputDataChannel :: TChan B.ByteString + ,_sessionThread :: ThreadId + ,_remainingOutputBit :: TMVar B.ByteString + ,_nh2Settings :: NH2.Settings + ,_sessionThrowed :: TMVar Bool + ,_waiting :: TMVar ThreadId + ,_encodeHeadersHere :: TMVar HP.DynamicTable + ,_decodeHeadersHere :: TMVar HP.DynamicTable + + ,_incomingData :: TMVar DataContinuityEngine + ,_clientState :: ClientState + -- Execute this when you want to use a powerfull client session + ,_startClientSessionCallback :: IO ClientState } L.makeLenses ''DecoySession + +-- errorsSessionConfig :: MVar Bool -> SessionsConfig +-- errorsSessionConfig mvar = set (sessionsCallbacks . reportErrorCallback_SC) +-- (Just $ setError mvar) defaultSessionsConfig + + +channelsToIOCallbacks :: TChan B.ByteString -> TChan B.ByteString -> TMVar DataContinuityEngine -> IOCallbacks +channelsToIOCallbacks input_data_channel output_data_channel incoming_data_tmvar = + let + push_action :: PushAction + push_action lazy_bs = atomically $ + writeTChan output_data_channel $ LB.toStrict lazy_bs + + pull_action :: PullAction + pull_action byte_count = atomically $ do + incoming_data <- takeTMVar incoming_data_tmvar + let + go dce = let + maybe_stuff = tryGetThisManyData dce byte_count + in case maybe_stuff of + Nothing -> do + new_data <- readTChan input_data_channel + let + new_dce = dceAddData dce new_data + go new_dce + + Just (new_bce, bs) -> do + putTMVar incoming_data_tmvar new_bce + return bs + go incoming_data + + best_effort_pull_action :: BestEffortPullAction + best_effort_pull_action can_block = atomically $ do + incoming_data <- takeTMVar incoming_data_tmvar + if dceIsEmpty incoming_data + then + if can_block + then do + -- Wait for the next item + new_data <- readTChan input_data_channel + -- Leave the empty thing alone + putTMVar incoming_data_tmvar incoming_data + return new_data + else do + putTMVar incoming_data_tmvar incoming_data + return "" + else do + let + (new_dce, all_together) = dceGetAllData incoming_data + putTMVar incoming_data_tmvar new_dce + return all_together + + close_action :: CloseAction + close_action = return () + in + IOCallbacks { + _pushAction_IOC = push_action, + _pullAction_IOC = pull_action, + _closeAction_IOC = close_action, + _bestEffortPullAction_IOC = best_effort_pull_action + } + + -- Supposed to be an HTTP/2 attendant createDecoySession :: Attendant -> IO DecoySession createDecoySession attendant = do - input_data_channel <- newChan - output_data_channel <- newChan + input_data_channel <- newTChanIO + output_data_channel <- newTChanIO + incoming_data_tmvar <- newTMVarIO emptyDCE + reverse_data_tmvar <- newTMVarIO emptyDCE + session_throwed_tmvar <- newTMVarIO False + client_state <- makeClientState + client_sessions_context <- makeSessionsContext defaultSessionsConfig let - push_action :: PushAction - push_action = writeChan output_data_channel - pull_action :: PullAction - pull_action = readChan input_data_channel - close_action :: CloseAction - best_effort_pull_action _ = error "NotImplemented" - close_action = return () - attendant_callbacks = AttendantCallbacks { - _pushAction_AtC = push_action, - _pullAction_AtC = pull_action, - _closeAction_AtC = close_action, - _bestEffortPullAction_AtC = best_effort_pull_action - } + attendant_callbacks = channelsToIOCallbacks input_data_channel output_data_channel incoming_data_tmvar + client_callbacks = channelsToIOCallbacks output_data_channel input_data_channel reverse_data_tmvar + start_client_session_callback :: IO ClientState + start_client_session_callback = do + wrapSession (ClientState_SP client_state) client_sessions_context client_callbacks + return client_state + dtable_for_encoding <- HP.newDynamicTableForEncoding 4096 - dtable_for_encoding_mvar <- newMVar dtable_for_encoding + dtable_for_encoding_mvar <- newTMVarIO dtable_for_encoding dtable_for_decoding <- HP.newDynamicTableForDecoding 4096 - dtable_for_decoding_mvar <- newMVar dtable_for_decoding + dtable_for_decoding_mvar <- newTMVarIO dtable_for_decoding - waiting_mvar <- newEmptyMVar + waiting_mvar <- newEmptyTMVarIO thread_id <- forkIO $ catch (do - attendant push_action pull_action close_action + attendant attendant_callbacks ) ((\ e -> do putStrLn $ "Exception: " ++ (show e) - maybe_waiting <- tryTakeMVar waiting_mvar + maybe_waiting <- atomically $ tryReadTMVar waiting_mvar case maybe_waiting of Just thread_id -> throwTo thread_id e Nothing -> return () ):: SomeException -> IO ()) - remaining_output_bit <- newMVar "" + remaining_output_bit <- newTMVarIO "" let session = DecoySession { _inputDataChannel = input_data_channel, _outputDataChannel = output_data_channel, @@ -86,7 +210,11 @@ _nh2Settings = NH2.defaultSettings, _waiting = waiting_mvar, _encodeHeadersHere = dtable_for_encoding_mvar, - _decodeHeadersHere = dtable_for_decoding_mvar + _decodeHeadersHere = dtable_for_decoding_mvar, + _incomingData = incoming_data_tmvar, + _sessionThrowed = session_throwed_tmvar, + _clientState = client_state, + _startClientSessionCallback = start_client_session_callback } return session @@ -97,13 +225,15 @@ -- Send a frame to a session -sendFrameToSession :: DecoySession -> OutputFrame -> IO () +sendFrameToSession :: DecoySession -> (NH2.EncodeInfo, NH2.FramePayload) -> IO () sendFrameToSession session (encode_info, frame_payload) = do let bs_list = NH2.encodeFrameChunks encode_info frame_payload input_data_channel = session ^. inputDataChannel - writeChan2Chan input_data_channel bs_list + atomically $ + mapM_ + (writeTChan input_data_channel) bs_list -- Read a frame from a session... if possible. It will block @@ -114,22 +244,30 @@ let output_data_channel = decoy_session ^. outputDataChannel remaining_output_bit_mvar = decoy_session ^. remainingOutputBit - pull_action = fmap LB.toStrict $ readChan output_data_channel + pull_action = readTChan output_data_channel settings = decoy_session ^. nh2Settings waiting_for_read = decoy_session ^. waiting my_thread_id <- myThreadId - putMVar waiting_for_read my_thread_id - remaining_output_bit <- takeMVar remaining_output_bit_mvar - (packet, rest) <- readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action - putMVar remaining_output_bit_mvar rest - takeMVar waiting_for_read + packet <- atomically $ do + {-# SCC aT #-} putTMVar waiting_for_read my_thread_id + remaining_output_bit <- {-# SCC bT #-} takeTMVar remaining_output_bit_mvar + (packet, rest) <- {-# SCC cT #-} readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action + {-# SCC dT #-} putTMVar remaining_output_bit_mvar rest + {-# SCC eT #-} takeTMVar waiting_for_read + return packet let error_or_frame = NH2.decodeFrame settings packet case error_or_frame of Left _ -> return Nothing Right frame -> return $ Just frame +-- Reads all frames from a session and assembles a response. Notice that this is a buggy hack, +-- because continuation frames and such are not considered. +-- T +-- readResponseFromSession :: DecoySession -> NH2.StreamId -> IO (Headers, B.ByteString, [NH2.Frame]) + + -- Encode headers to send to the session -- TODO: There is an important bug here... we are using the default encoding -- strategy everywhere @@ -138,21 +276,20 @@ do let mv = decoy_session ^. encodeHeadersHere - dtable <- takeMVar mv + dtable <- atomically $ takeTMVar mv (dtable', bs ) <- HP.encodeHeader HP.defaultEncodeStrategy dtable headers - putMVar mv dtable' + atomically $ putTMVar mv dtable' return bs -- Decode headers to receive them from the session decodeHeadersForSession :: DecoySession -> B.ByteString -> IO Headers -decodeHeadersForSession decoy_session bs = - do - let - mv = decoy_session ^. decodeHeadersHere - dtable <- takeMVar mv - (dtable', headers) <- HP.decodeHeader dtable bs - putMVar mv dtable' - return headers +decodeHeadersForSession decoy_session bs = do + let + mv = decoy_session ^. decodeHeadersHere + dtable <- atomically $ takeTMVar mv + (dtable', headers) <- HP.decodeHeader dtable bs + atomically $ putTMVar mv dtable' + return headers -- Send raw data to a session sendRawDataToSession :: DecoySession -> B.ByteString -> IO () @@ -162,9 +299,10 @@ waiting_for_write = decoy_session ^. waiting thread_id <- myThreadId - putMVar waiting_for_write thread_id - writeChan input_data_channel data_to_send - takeMVar waiting_for_write + atomically $ do + putTMVar waiting_for_write thread_id + writeTChan input_data_channel data_to_send + takeTMVar waiting_for_write return () diff --git a/tests/tests-hs-src/Tests/HTTP1Parse.hs b/tests/tests-hs-src/Tests/HTTP1Parse.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/Tests/HTTP1Parse.hs @@ -0,0 +1,46 @@ +{-# LANGUAGE OverloadedStrings #-} +module Tests.HTTP1Parse where + + +import qualified Data.ByteString as B +import Data.Maybe (isJust) + +import Test.HUnit + +import SecondTransfer (Headers) +import SecondTransfer.Http1.Parse + + +testParse :: Test +testParse = TestCase $ do + let + headers_text = "GET /helo.html HTTP/1.1\r\nHost: www.auther.com \r\n\r\n" + headers_text2 = "POST /helo.html HTTP/1.1\r\nHost: www.auther.com \r\nContent-Length: 1\r\n\r\np" + a0 = newIncrementalHttp1Parser + isDone (OnlyHeaders_H1PC _ _) = True + isDone _ = False + a1 = addBytes a0 headers_text + (OnlyHeaders_H1PC h0 leftovers)= a1 + (HeadersAndBody_H1PC h1 cond0 l2) = addBytes a0 headers_text2 + waitForBodyOk (HeadersAndBody_H1PC _ _ _) = True + waitForBodyOk _ = False + assertBool "testParse.IsDone" (isDone a1) + assertEqual "testParse.NoLeftovers" leftovers "" + assertEqual "testParse.YesLeftovers" l2 "p" + assertEqual "testParse.FinishWellSeen" (UseBodyLength_BSC 1) cond0 + + +testGenerate :: Test +testGenerate = TestCase $ do + let + headers_list :: Headers + headers_list = [ + (":status", "200"), + ("host", "www.example.com"), + ("etag", "afrh") + ] + fragments = ["hello world"] + serialized = serializeHTTPResponse headers_list fragments + assertEqual "testGenerate.1" + "HTTP/1.1 200 OK\r\ncontent-length: 11\r\netag: afrh\r\nhost: www.example.com\r\n\r\nhello world" + serialized diff --git a/tests/tests-hs-src/Tests/HTTP2Session.hs b/tests/tests-hs-src/Tests/HTTP2Session.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/Tests/HTTP2Session.hs @@ -0,0 +1,584 @@ +{-# LANGUAGE OverloadedStrings #-} +module Tests.HTTP2Session where + + +import Data.Typeable + +import Control.Concurrent (threadDelay) +import qualified Control.Concurrent as C (yield) +import Control.Concurrent.MVar +import Control.Exception +import Control.Lens +import qualified Control.Lens as L +import Control.Monad.IO.Class (liftIO) +import qualified Network.HTTP2 as NH2 +import Test.HUnit +import SecondTransfer.Exception +import SecondTransfer.Http2 (http2Attendant) +import SecondTransfer.Sessions +import SecondTransfer.Test.DecoySession +import SecondTransfer.Types +import SecondTransfer.Utils.HTTPHeaders (fetchHeader) +import SecondTransfer.MainLoop.CoherentWorker (defaultEffects) +import SecondTransfer.MainLoop.ClientPetitioner + + + +import Data.Conduit (yield) + + + +saysHello :: DataAndConclusion +saysHello = do + yield "Hello world!\ns" + -- No footers + return [] + + +simpleWorker :: AwareWorker +simpleWorker = coherentToAwareWorker . const $ return ( + [ + (":status", "200") + ], + [], -- No pushed streams + saysHello + ) + + +erringWorker :: AwareWorker +erringWorker = coherentToAwareWorker . const $ return ( + [ + (":status", "500") + ], + [], -- No pushed streams + saysHello + ) + + +abortingWorker :: AwareWorker +abortingWorker req_ = do + pr1 <- erringWorker req_ + let + pr2 = L.set (effect_PS . interrupt_Ef) (Just InterruptConnectionAfter_IEf) pr1 + return pr2 + + +earlyAbortingWorker :: AwareWorker +earlyAbortingWorker req_ = do + pr1 <- erringWorker req_ + let + pr2 = L.set (effect_PS . interrupt_Ef) (Just InterruptConnectionNow_IEf) pr1 + return pr2 + + +data Internal500Exception = Internal500Exception + deriving (Typeable, Show) + +instance Exception Internal500Exception where + toException = convertHTTP500PrecursorExceptionToException + fromException = getHTTP500PrecursorExceptionFromException + + +throwingWorker :: AwareWorker +throwingWorker _ = throwIO Internal500Exception + +throwingWorker2 :: AwareWorker +throwingWorker2 = coherentToAwareWorker . const . return $ ( + [ + -- These headers will be already sent by the time + -- the exception is discovered... + (":status", "200") + ], + [], -- No pushed streams + do + yield "Error coming down" + liftIO $ throwIO Internal500Exception + return [] + ) + + +simpleRequestHeaders :: Headers +simpleRequestHeaders = [ + (":path", "/"), + (":authority", "www.example.com"), + (":scheme", "https"), + (":method", "GET") + ] + + +badRequestHeaders :: Headers +badRequestHeaders = [ + (":path", "/"), + (":authority", "www.example.com"), + (":scheme", "https") + ] + + +setError :: MVar Bool -> ErrorCallback +setError mvar = const $ modifyMVar_ mvar (const $ return True ) + + +errorForPrefaceOk :: MVar Bool -> ErrorCallback +errorForPrefaceOk ok_mvar (_, _, some_exception) = do + let + maybe_blocked :: Maybe BlockedIndefinitelyOnMVar + maybe_blocked = fromException some_exception + case maybe_blocked of + Just _ -> modifyMVar_ ok_mvar (const . return $ True) + Nothing -> modifyMVar_ ok_mvar (const . return $ False) + + +errorsSessionConfig :: MVar Bool -> SessionsConfig +errorsSessionConfig mvar = set (sessionsCallbacks . reportErrorCallback_SC) + (Just $ setError mvar) defaultSessionsConfig + +errorsSessionConfigForMVar :: MVar Bool -> SessionsConfig +errorsSessionConfigForMVar mvar = set (sessionsCallbacks . reportErrorCallback_SC) + (Just $ errorForPrefaceOk mvar) defaultSessionsConfig + + +-- Test disabled due to irrelevance +-- testPrefaceChecks :: Test +-- testPrefaceChecks = TestCase $ do +-- errors_mvar <- newMVar False +-- sessions_context <- makeSessionsContext (errorsSessionConfigForMVar errors_mvar) +-- let +-- attendant = http2Attendant sessions_context simpleWorker +-- decoy_session <- createDecoySession attendant +-- -- This should work +-- sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" +-- threadDelay 1000000 +-- error_ok <- readMVar errors_mvar +-- if not error_ok then +-- assertFailure "TypeOfErrorNotOK" +-- else +-- return () + + +testPrefaceChecks2 :: Test +testPrefaceChecks2 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + -- This should work + sendRawDataToSession decoy_session "PRI * HXXP/2.0\r\n\r\nSM\r\n\r\n" + threadDelay 1000000 + got_error <- readMVar errors_mvar + if not got_error then do + assertFailure "Exception didn't raise" + else + return () + + +testFirstFrameMustBeSettings :: Test +testFirstFrameMustBeSettings = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + maybe_frame <- recvFrameFromSession decoy_session + case maybe_frame of + Nothing -> + assertFailure "Waiting a frame, received none" + Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok + return () + + _ -> + assertFailure "Waiting a settings frame, received something else" + + +testFirstFrameMustBeSettings2 :: Test +testFirstFrameMustBeSettings2 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + maybe_frame <- recvFrameFromSession decoy_session + case maybe_frame of + Nothing -> + assertFailure "Waiting a frame, received none" + Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok + return () + + _ -> + assertFailure "Waiting a settings frame, received something else" + + -- Send a settings frame now + sendFrameToSession + decoy_session + ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing), + (NH2.SettingsFrame []) + ) + + -- Check session is alive + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + +testFirstFrameMustBeSettings3 :: Test +testFirstFrameMustBeSettings3 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + + let d2 = ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing), + (NH2.PingFrame "01234567") ) + + -- Send a ping frame now, so that we get an error + sendFrameToSession + decoy_session + d2 + + -- We need to give some time to the framework to react to problems + threadDelay 20000 + + -- Check session is alive + got_error <- readMVar errors_mvar + if not got_error then do + assertFailure "Exception didn't raise properly" + else + return () + + +testIGet500Status :: Test +testIGet500Status = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context throwingWorker + decoy_session <- createDecoySession attendant + + -- Send the prologue + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + + -- Send a settings frame now, otherwise the session will bark.... + sendFrameToSession + decoy_session + ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing), + (NH2.SettingsFrame []) ) + + -- Now perform a simple, mocking request + performRequestSimple decoy_session 1 [ + (":method", "get"), + (":scheme", "https"), + (":authority", "www.example.com"), + (":path", "/hi") + ] + + -- Now we read a few frames + seen <- return False + f0 <- recvFrameFromSession decoy_session + seen1 <- frameIsStatus500 decoy_session seen f0 + + f1 <- recvFrameFromSession decoy_session + seen2 <- frameIsStatus500 decoy_session seen f1 + + f2 <- recvFrameFromSession decoy_session + seen3 <- frameIsStatus500 decoy_session seen f2 + + if not seen3 then do + assertFailure "Didn't see that 500" + else + return () + + +frameIsStatus500 :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool +frameIsStatus500 decoy_session prev maybe_frame = + case prev of + True -> return True + False -> + case maybe_frame of + Just (NH2.Frame _ (NH2.HeadersFrame _ bs ) ) -> do + headers <- decodeHeadersForSession decoy_session bs + let + maybe_status = fetchHeader headers ":status" + case maybe_status of + Just x | x == "500" -> return True + _ -> return False + + _ -> + return False + + +testSessionBreaksOnLateError :: Test +testSessionBreaksOnLateError = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context throwingWorker2 + decoy_session <- createDecoySession attendant + + -- Send the prologue + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + + -- Send a settings frame now, otherwise the session will bark.... + sendFrameToSession + decoy_session + ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing), + (NH2.SettingsFrame []) ) + + -- Now perform a simple, mocking request + performRequestSimple decoy_session 1 [ + (":method", "get"), + (":scheme", "https"), + (":authority", "www.example.com"), + (":path", "/hi") + ] + + -- Now we read a few frames + seen <- return False + f0 <- recvFrameFromSession decoy_session + seen1 <- frameIsGoAwayBecauseInternalError decoy_session seen f0 + + f1 <- recvFrameFromSession decoy_session + seen2 <- frameIsGoAwayBecauseInternalError decoy_session seen1 f1 + + f2 <- recvFrameFromSession decoy_session + seen3 <- frameIsGoAwayBecauseInternalError decoy_session seen2 f2 + + f3 <- recvFrameFromSession decoy_session + seen4 <- frameIsGoAwayBecauseInternalError decoy_session seen3 f3 + + if not seen4 then do + assertFailure "Didn't see GoAwayFrame" + else + return () + + +frameIsGoAwayBecauseInternalError :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool +frameIsGoAwayBecauseInternalError decoy_session prev maybe_frame = do + case prev of + True -> return True + False -> + case maybe_frame of + Just (NH2.Frame _ (NH2.GoAwayFrame _ ec _) ) -> do + case ec of + NH2.InternalError -> return True + _ -> return False + + _ -> + return False + + +frameIsGoAwayBecauseProtocolError :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool +frameIsGoAwayBecauseProtocolError decoy_session prev maybe_frame = do + case prev of + True -> return True + False -> + case maybe_frame of + Just (NH2.Frame _ (NH2.GoAwayFrame _ ec _) ) -> do + case ec of + NH2.ProtocolError -> return True + _ -> return False + + _ -> + return False + + +testUpdateWindowFrameAborts :: Test +testUpdateWindowFrameAborts = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + maybe_frame <- recvFrameFromSession decoy_session + case maybe_frame of + Nothing -> + assertFailure "Waiting a frame, received none" + Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok + return () + + _ -> + assertFailure "Waiting a settings frame, received something else" + + -- Send a settings frame now + sendFrameToSession + decoy_session + ( NH2.EncodeInfo NH2.defaultFlags 0 Nothing, + NH2.SettingsFrame [] ) + + -- And now send a WindowUpdate frame + sendFrameToSession + decoy_session + (NH2.EncodeInfo NH2.defaultFlags 51 Nothing, + NH2.WindowUpdateFrame 10 ) + + -- Now we read a few frames + seen <- return False + f0 <- recvFrameFromSession decoy_session + seen1 <- frameIsGoAwayBecauseProtocolError decoy_session seen f0 + + f1 <- recvFrameFromSession decoy_session + seen2 <- frameIsGoAwayBecauseInternalError decoy_session seen1 f1 + + -- f2 <- recvFrameFromSession decoy_session + -- seen3 <- frameIsGoAwayBecauseProtocolError decoy_session seen2 f2 + + -- f3 <- recvFrameFromSession decoy_session + -- seen4 <- frameIsGoAwayBecauseProtocolError decoy_session seen3 f3 + + if not seen2 then do + assertFailure "Didn't see GoAwayFrame" + else + return () + + -- Check session is alive + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + +testClosedInteraction0 :: Test +testClosedInteraction0 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + let + start_client = decoy_session ^. startClientSessionCallback + start_client + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + + +testClosedInteraction1 :: Test +testClosedInteraction1 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + let + start_client = decoy_session ^. startClientSessionCallback + client_state <- start_client + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + + (headers, input_data_stream) <- request client_state simpleRequestHeaders (return ()) + if length headers <= 0 + then + assertFailure "NoHeadersBack" + else + return () + return () + + + +testClosedInteraction3 :: Test +testClosedInteraction3 = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context simpleWorker + decoy_session <- createDecoySession attendant + let + start_client = decoy_session ^. startClientSessionCallback + client_state <- start_client + ee_mvar <- newMVar False + catch + (do + request client_state badRequestHeaders (return ()) + return () + ) + ((\ _ -> modifyMVar_ ee_mvar ( \ _ -> return $ True) ):: ClientSessionAbortedException -> IO () ) + -- if got_error then do + -- return () + -- else + -- assertFailure "IWasExpectingAnError" + ee <- takeMVar ee_mvar + if ee + then + return () + else + assertFailure "IWasExpectingAnError--" + + + +testWorkerClosesAfter :: Test +testWorkerClosesAfter = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context abortingWorker + decoy_session <- createDecoySession attendant + let + start_client = decoy_session ^. startClientSessionCallback + client_state <- start_client + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + + (headers, _) <- request client_state simpleRequestHeaders (return ()) + if length headers <= 0 + then + assertFailure "NoHeadersBack" + else + return () + + -- Now, when I try to connect, something bad should happen, since the server must + -- be closed + + got_error2 <- newMVar False + catch + (request client_state simpleRequestHeaders (return ()) >> return () ) + ((\ _ -> modifyMVar_ got_error2 ( \ _ -> return $ True) )::ClientSessionAbortedException -> IO () ) + ee <- readMVar got_error2 + if ee + then + return () + else + assertFailure "Server Must Be Closed" + + return () + +testWorkerClosesBefore :: Test +testWorkerClosesBefore = TestCase $ do + errors_mvar <- newMVar False + sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar) + let + attendant = http2Attendant sessions_context earlyAbortingWorker + decoy_session <- createDecoySession attendant + let + start_client = decoy_session ^. startClientSessionCallback + client_state <- start_client + got_error <- readMVar errors_mvar + if got_error then do + assertFailure "Exception raised unexpectedly" + else + return () + + got_error2 <- newMVar False + catch + (request client_state simpleRequestHeaders (return ()) >> return () ) + ((\ _ -> modifyMVar_ got_error2 ( \ _ -> return $ True) )::ClientSessionAbortedException -> IO () ) + ee <- readMVar got_error2 + if ee + then + return () + else + assertFailure "Server Must Be Closed" + + return () diff --git a/tests/tests-hs-src/Tests/Utils.hs b/tests/tests-hs-src/Tests/Utils.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/Tests/Utils.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE OverloadedStrings #-} +module Tests.Utils where + + +import qualified Data.ByteString as B + +import Test.HUnit + +import SecondTransfer (Headers) +import SecondTransfer.Utils.HTTPHeaders +import SecondTransfer.Http1.Parse (locateCRLFs) + + +testLowercaseHeaders :: Test +testLowercaseHeaders = TestCase $ do + let + irregular = [ + ("A", "B"), + ("C-D", "D") + ] :: Headers + regular = lowercaseHeaders irregular + assertEqual "lowercase" regular [ + ("a", "B"), + ("c-d", "D") + ] + assertEqual "test-lowercase-1" (headersAreLowercase irregular) False + assertEqual "test-lowercase-2" (headersAreLowercase regular) True + + +testCRLFLocate :: Test +testCRLFLocate = TestCase $ do + let + input_text = "hello\r\nworld" + (positions, length_, lastchar) = locateCRLFs 0 [] 0 input_text + assertEqual "position-is-5.position" (head positions) 5 + assertEqual "position-is-5.length" length_ (B.length input_text) + assertEqual "position-is-5.lastchar" lastchar (fromIntegral . fromEnum $ 'd') + + + let + input_text_2 = "hello\r\nworld\r\n\r\n" + (positions, length_, lastchar) = locateCRLFs 0 [] 0 input_text_2 + assertEqual "many-positions.position" positions [14,12,5] + assertEqual "many-positions.length" length_ (B.length input_text_2) + assertEqual "many-positions.lastchar" lastchar (fromIntegral . fromEnum $ '\n') + + +testReplaceHostByAuthority :: Test +testReplaceHostByAuthority = TestCase $ do + let + headers_pre = [ + ("A", "B"), + ("C-D", "D"), + ("Host", "D") + ] :: Headers + regular = toList. replaceHostByAuthority . fromList . lowercaseHeaders $ headers_pre + -- Do not pass in headers with uppercase + assertEqual "replace-host-by-authority" + [ + (":authority", "D"), + ("a", "B"), + ("c-d", "D") + ] + regular diff --git a/tests/tests-hs-src/hunit_tests.hs b/tests/tests-hs-src/hunit_tests.hs --- a/tests/tests-hs-src/hunit_tests.hs +++ b/tests/tests-hs-src/hunit_tests.hs @@ -14,24 +14,29 @@ tests = TestList [ - TestLabel "testPrefaceChecks" testPrefaceChecks, - TestLabel "testPrefaceChecks2" testPrefaceChecks2, - TestLabel "testLowercaseHeaders" testLowercaseHeaders, - TestLabel "testCRLFLocate" testCRLFLocate, - TestLabel "testHTTP1Parse" testParse, - TestLabel "testGenerate" testGenerate, - TestLabel "testReplaceHostByAuthority" testReplaceHostByAuthority, - TestLabel "testFirstFrameIsSettings" testFirstFrameMustBeSettings, - TestLabel "testFirstFrameIsSettings2" testFirstFrameMustBeSettings2, - TestLabel "testFirstFrameIsSettings3" testFirstFrameMustBeSettings3, - TestLabel "testIGet500Status" testIGet500Status, - TestLabel "testSessionBreaksOnLateError" testSessionBreaksOnLateError, - TestLabel "WindowUpdate to unexistent stream" testUpdateWindowFrameAborts + --TestLabel "testPrefaceChecks" testPrefaceChecks + TestLabel "testPrefaceChecks2" testPrefaceChecks2 + ,TestLabel "testLowercaseHeaders" testLowercaseHeaders + ,TestLabel "testCRLFLocate" testCRLFLocate + ,TestLabel "testHTTP1Parse" testParse + ,TestLabel "testGenerate" testGenerate + ,TestLabel "testReplaceHostByAuthority" testReplaceHostByAuthority + ,TestLabel "testFirstFrameIsSettings" testFirstFrameMustBeSettings + ,TestLabel "testFirstFrameIsSettings2" testFirstFrameMustBeSettings2 + ,TestLabel "testFirstFrameIsSettings3" testFirstFrameMustBeSettings3 + ,TestLabel "testIGet500Status" testIGet500Status + ,TestLabel "testSessionBreaksOnLateError" testSessionBreaksOnLateError + ,TestLabel "WindowUpdate to unexistent stream" testUpdateWindowFrameAborts + ,TestLabel "testClosedInteraction0" testClosedInteraction0 + ,TestLabel "testClosedInteraction1" testClosedInteraction1 + ,TestLabel "testClosedInteraction3" testClosedInteraction3 + ,TestLabel "testWorkerClosesAfter" testWorkerClosesAfter + ,TestLabel "testWorkerClosesBefore" testWorkerClosesBefore ] -main = do - rets <- runTestTT tests +main = do + rets <- runTestTT tests if (errors rets == 0 && failures rets == 0) then exitWith ExitSuccess else exitWith $ ExitFailure (-1)