second-transfer-0.6.0.0: hs-src/SecondTransfer/Http2/Session.hs
-- Session: links frames to streams, and helps in ordering the header frames
-- so that they don't get mixed with header frames from other streams when
-- resources are being served concurrently.
{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}
{-# OPTIONS_HADDOCK hide #-}
module SecondTransfer.Http2.Session(
http2Session
,getFrameFromSession
,sendFirstFrameToSession
,sendMiddleFrameToSession
,sendCommandToSession
,CoherentSession
,SessionInput(..)
,SessionInputCommand(..)
,SessionOutput(..)
,SessionOutputCommand(..)
,SessionCoordinates(..)
,SessionComponent(..)
,SessionsCallbacks(..)
,SessionsConfig(..)
,ErrorCallback
-- Internal stuff
,OutputFrame
,InputFrame
) where
#include "Logging.cpphs"
-- System grade utilities
import Control.Concurrent (ThreadId, forkIO, threadDelay)
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 qualified Data.ByteString as B
import Data.ByteString.Char8 (pack,unpack)
import qualified Data.ByteString.Builder as Bu
import qualified Data.ByteString.Lazy as Bl
import Data.Conduit
import qualified Data.HashTable.IO as H
import qualified Data.IntSet as NS
import Data.Maybe (isJust)
#ifndef IMPLICIT_MONOID
import Data.Monoid (mappend)
#endif
import Control.Lens
-- No framing layer here... let's use Kazu's Yamamoto library
import qualified Network.HPACK as HP
import qualified Network.HTTP2 as NH2
-- Logging utilities
import System.Log.Logger
import System.Clock (getTime, TimeSpec, Clock(..))
-- Imports from other parts of the program
import SecondTransfer.MainLoop.CoherentWorker
import SecondTransfer.MainLoop.Tokens
import SecondTransfer.Sessions.Config
import SecondTransfer.Sessions.Internal (sessionExceptionHandler,
SessionsContext,
sessionsConfig)
import SecondTransfer.Utils (unfoldChannelAndSource)
import SecondTransfer.Exception
import qualified SecondTransfer.Utils.HTTPHeaders as He
import SecondTransfer.MainLoop.Logging (logWithExclusivity, logit)
-- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to
-- use :-(
type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload, Effect)
type InputFrame = NH2.Frame
--useChunkLength :: Int
-- useChunkLength = 2048
-- Singleton instance used for concurrency
data HeadersSent = HeadersSent
-- All streams put their data bits here. A "Nothing" value signals
-- end of data. Middle value is delay in microseconds
type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString, Effect)
data HeaderOutputMessage =
NormalResponse_HM (GlobalStreamId, MVar HeadersSent, Headers, Effect)
--
|PushPromise_HM (GlobalStreamId, GlobalStreamId, Headers, Effect)
-- Settings imposed by the peer
data SessionSettings = SessionSettings {
_pushEnabled :: Bool
}
makeLenses ''SessionSettings
-- Whatever a worker thread is going to need comes here....
-- this is to make refactoring easier, but not strictly needed.
data WorkerThreadEnvironment = WorkerThreadEnvironment {
-- What's the header stream id?
_streamId :: GlobalStreamId
-- A full block of headers can come here... the mvar in the middle should
-- be populate to signal end of headers transmission. A thread will be suspended
-- waiting for that
, _headersOutput :: Chan HeaderOutputMessage
-- And regular contents can come this way and thus be properly mixed
-- with everything else.... for now...
,_dataOutput :: MVar DataOutputToConveyor
,_streamsCancelled_WTE :: MVar NS.IntSet
,_sessionSettings_WTE :: MVar SessionSettings
,_nextPushStream_WTE :: MVar Int
}
makeLenses ''WorkerThreadEnvironment
-- An HTTP/2 session. Basically a couple of channels ...
type Session = (SessionInput, SessionOutput)
-- From outside, one can only write to this one ... the newtype is to enforce
-- this.
newtype SessionInput = SessionInput ( Chan SessionInputCommand )
sendMiddleFrameToSession :: SessionInput -> InputFrame -> IO ()
sendMiddleFrameToSession (SessionInput chan) frame = writeChan chan $ MiddleFrame_SIC frame
sendFirstFrameToSession :: SessionInput -> InputFrame -> IO ()
sendFirstFrameToSession (SessionInput chan) frame = writeChan chan $ FirstFrame_SIC frame
sendCommandToSession :: SessionInput -> SessionInputCommand -> IO ()
sendCommandToSession (SessionInput chan) command = writeChan chan command
type SessionOutputPacket = Either SessionOutputCommand OutputFrame
type SessionOutputChannelAbstraction = Chan SessionOutputPacket
-- From outside, one can only read from this one
newtype SessionOutput = SessionOutput SessionOutputChannelAbstraction
getFrameFromSession :: SessionOutput -> IO SessionOutputPacket
getFrameFromSession (SessionOutput chan) = readChan chan
type HashTable k v = H.CuckooHashTable k v
type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder
type WorkerMonad = ReaderT WorkerThreadEnvironment IO
-- Have to figure out which are these...but I would expect to have things
-- like unexpected aborts here in this type.
data SessionInputCommand =
FirstFrame_SIC InputFrame -- This frame is special
|MiddleFrame_SIC InputFrame -- Ordinary frame
|InternalAbort_SIC -- Internal abort from the session itself
|CancelSession_SIC -- Cancel request from the framer
deriving Show
-- temporary
data SessionOutputCommand =
CancelSession_SOC
|FinishStream_SOC GlobalStreamId
deriving Show
-- Here is how we make a session
type SessionMaker = SessionsContext -> IO Session
-- Here is how we make a session wrapping a CoherentWorker
type CoherentSession = AwareWorker -> SessionMaker
data PostInputMechanism = PostInputMechanism (MVar (Maybe B.ByteString), InputDataStream)
-- NH2.Frame != Frame
data SessionData = SessionData {
-- ATTENTION: Ignore the warning coming from here for now
_sessionsContext :: SessionsContext
,_sessionInput :: Chan SessionInputCommand
-- We need to lock this channel occassionally so that we can order multiple
-- header frames properly....that's the reason for the outer MVar
,_sessionOutput :: MVar SessionOutputChannelAbstraction
-- Use to encode
,_toEncodeHeaders :: MVar HP.DynamicTable
-- And used to decode
,_toDecodeHeaders :: MVar HP.DynamicTable
-- While I'm receiving headers, anything which
-- is not a header should end in the connection being
-- closed
,_receivingHeaders :: MVar (Maybe Int)
-- _lastGoodStream is used both to report the last good stream in the
-- GoAwayFrame and to keep track of streams oppened by the client. In
-- other words, it contains the stream_id of the last valid client
-- stream and is updated as soon as the first frame of that stream is
-- received.
,_lastGoodStream :: MVar Int
-- Used for decoding the headers
,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment
-- Used for worker threads... this is actually a pre-filled template
-- I make copies of it in different contexts, and as needed.
,_forWorkerThread :: WorkerThreadEnvironment
,_awareWorker :: AwareWorker
-- Some streams may be cancelled
,_streamsCancelled :: MVar NS.IntSet
-- Data input mechanism corresponding to some threads
,_stream2PostInputMechanism :: HashTable Int PostInputMechanism
-- Worker thread register. This is a dictionary from stream id to
-- the ThreadId of the thread with the worker thread. I use this to
-- raise asynchronous exceptions in the worker thread if the stream
-- is cancelled by the client. This way we get early finalization.
,_stream2WorkerThread :: HashTable Int ThreadId
-- Use to retrieve/set the session id
,_sessionIdAtSession :: Int
-- And used to keep peer session settings
,_sessionSettings :: MVar SessionSettings
-- What is the next stream available for push?
,_nextPushStream :: MVar Int
}
makeLenses ''SessionData
-- v- {headers table size comes here!!}
http2Session :: AwareWorker -> Int -> SessionsContext -> IO Session
http2Session aware_worker session_id sessions_context = do
session_input <- newChan
session_output <- newChan
session_output_mvar <- newMVar session_output
-- For incremental construction of headers...
stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment
-- Warning: we should find a way of coping with different table sizes.
decode_headers_table <- HP.newDynamicTableForDecoding 4096
decode_headers_table_mvar <- newMVar decode_headers_table
encode_headers_table <- HP.newDynamicTableForEncoding 4096
encode_headers_table_mvar <- newMVar encode_headers_table
-- These ones need independent threads taking care of sending stuff
-- their way...
headers_output <- newChan :: IO (Chan HeaderOutputMessage)
data_output <- newEmptyMVar :: IO (MVar DataOutputToConveyor)
stream2postinputmechanism <- H.new
stream2workerthread <- H.new
last_good_stream_mvar <- newMVar (-1)
receiving_headers <- newMVar Nothing
session_settings <- newMVar $ SessionSettings { _pushEnabled = True }
next_push_stream <- newMVar 8
-- What about stream cancellation?
cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet)
let for_worker_thread = WorkerThreadEnvironment {
_streamId = error "NotInitialized"
,_headersOutput = headers_output
,_dataOutput = data_output
,_streamsCancelled_WTE = cancelled_streams_mvar
,_sessionSettings_WTE = session_settings
,_nextPushStream_WTE = next_push_stream
}
let session_data = SessionData {
_sessionsContext = sessions_context
,_sessionInput = session_input
,_sessionOutput = session_output_mvar
,_toDecodeHeaders = decode_headers_table_mvar
,_toEncodeHeaders = encode_headers_table_mvar
,_stream2HeaderBlockFragment = stream_request_headers
,_forWorkerThread = for_worker_thread
,_awareWorker = aware_worker
,_streamsCancelled = cancelled_streams_mvar
,_stream2PostInputMechanism = stream2postinputmechanism
,_stream2WorkerThread = stream2workerthread
,_sessionIdAtSession = session_id
,_receivingHeaders = receiving_headers
,_sessionSettings = session_settings
,_lastGoodStream = last_good_stream_mvar
,_nextPushStream = next_push_stream
}
let
exc_handler :: SessionComponent -> HTTP2SessionException -> IO ()
exc_handler component e = sessionExceptionHandler component session_id sessions_context e
exc_guard :: SessionComponent -> IO () -> IO ()
exc_guard component action = E.catch
action
(\e -> do
-- INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" )
exc_handler component e
)
use_chunk_length = sessions_context ^. sessionsConfig . dataFrameSize
-- Create an input thread that decodes frames...
forkIO $ exc_guard SessionInputThread_HTTP2SessionComponent
$ runReaderT sessionInputThread session_data
-- Create a thread that captures headers and sends them down the tube
forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent
$ runReaderT (headersOutputThread headers_output session_output_mvar) session_data
-- Create a thread that captures data and sends it down the tube. This is waiting for
-- stuff coming from all the workers. This function is also in charge of closing streams
forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent
$ dataOutputThread use_chunk_length data_output session_output_mvar
-- The two previous threads fill the session_output argument below (they write to it)
-- the session machinery in the other end is in charge of sending that data through the
-- socket.
return ( (SessionInput session_input),
(SessionOutput session_output) )
-- TODO: Some ill clients can break this thread with exceptions. Make these paths a bit
--- more robust.
sessionInputThread :: ReaderT SessionData IO ()
sessionInputThread = do
INSTRUMENTATION( debugM "HTTP2.Session" "Entering sessionInputThread" )
-- This is an introductory and declarative block... all of this is tail-executed
-- every time that a packet needs to be processed. It may be a good idea to abstract
-- these values in a closure...
session_input <- view sessionInput
decode_headers_table_mvar <- view toDecodeHeaders
stream_request_headers <- view stream2HeaderBlockFragment
cancelled_streams_mvar <- view streamsCancelled
coherent_worker <- view awareWorker
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
input <- liftIO $ readChan session_input
case input of
FirstFrame_SIC (NH2.Frame
(NH2.FrameHeader _ 1 null_stream_id ) _ )| 0 == null_stream_id -> do
-- This is a SETTINGS ACK frame, which is okej to have,
-- do nothing here
continue
FirstFrame_SIC
(NH2.Frame
(NH2.FrameHeader _ 0 null_stream_id )
(NH2.SettingsFrame settings_list)
) | 0 == null_stream_id -> do
-- Good, handle
handleSettingsFrame settings_list
continue
FirstFrame_SIC _ -> do
-- Bad, incorrect id or god knows only what ....
closeConnectionBecauseIsInvalid NH2.ProtocolError
return ()
CancelSession_SIC -> do
-- Good place to tear down worker threads... Let the rest of the finalization
-- to the framer.
--
-- This message is normally got from the Framer
liftIO $ do
H.mapM_
(\ (_, thread_id) -> do
throwTo thread_id StreamCancelledException
return ()
)
stream2workerthread
-- We do not continue here, but instead let it finish
return ()
InternalAbort_SIC -> do
-- Message triggered because the worker failed to behave.
-- When this is sent, the connection is closed
closeConnectionBecauseIsInvalid NH2.InternalError
return ()
-- The block below will process both HEADERS and CONTINUATION frames.
-- TODO: As it stands now, the server will happily start a new stream with
-- a CONTINUATION frame instead of a HEADERS frame. That's against the
-- protocol.
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
}
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 ()
continue
MiddleFrame_SIC frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do
let stream_id = streamIdFromFrame frame
liftIO $ do
-- INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show _error_code_id) )
cancelled_streams <- takeMVar cancelled_streams_mvar
-- INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) )
putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams
maybe_thread_id <- H.lookup stream2workerthread stream_id
case maybe_thread_id of
Nothing ->
-- This is actually more like an internal error, when this
-- happens, cancel the session
error "InterruptingUnexistentStream"
Just thread_id -> do
-- INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" )
throwTo thread_id StreamCancelledException
continue
MiddleFrame_SIC frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes))
-> unlessReceivingHeaders $ do
-- So I got data to process
-- TODO: Handle end of stream
let stream_id = nh2_stream_id
-- TODO: Handle the cases where the stream_id doesn't match an already existent
-- stream. In such cases it is justified to reset the connection with a protocol_error.
streamWorkerSendData stream_id somebytes
-- After that data has been received and forwarded downstream, we can issue a windows update
--
-- TODO: We can use wider frames to avoid congestion...
-- .... and we can also be more compositional with these short bursts of data....
--
-- TODO: Consider that the best place to output these frames can be somewhere else...
--
-- TODO: Use a special, with-quota queue here to do flow control. Don't send meaningless
-- WindowUpdateFrame's
sendOutFrame
(NH2.EncodeInfo
NH2.defaultFlags
nh2_stream_id
Nothing
)
(NH2.WindowUpdateFrame
(fromIntegral (B.length somebytes))
)
sendOutFrame
(NH2.EncodeInfo
NH2.defaultFlags
0
Nothing
)
(NH2.WindowUpdateFrame
(fromIntegral (B.length somebytes))
)
if frameEndsStream frame
then do
-- Good place to close the source ...
closePostDataSource stream_id
else
return ()
continue
MiddleFrame_SIC (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do
-- Deal with pings: this is an Ack, so do nothing
continue
MiddleFrame_SIC (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do
-- Deal with pings: NOT an Ack, so answer
-- INSTRUMENTATION( debugM "HTTP2.Session" "Ping processed" )
sendOutFrame
(NH2.EncodeInfo
(NH2.setAck NH2.defaultFlags)
0
Nothing
)
(NH2.PingFrame somebytes)
continue
MiddleFrame_SIC (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do
-- Frame was received by the peer, do nothing here...
continue
-- TODO: Do something with these settings!!
MiddleFrame_SIC (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do
-- INSTRUMENTATION( debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) )
-- Just acknowledge the frame.... for now
handleSettingsFrame settings_list
continue
MiddleFrame_SIC somethingelse -> unlessReceivingHeaders $ do
-- An undhandled case here....
-- liftIO . logit $ "Strange frame:" `mappend` (pack . show $ somethingelse)
continue
where
continue = sessionInputThread
-- TODO: Do use the settings!!!
handleSettingsFrame :: NH2.SettingsList -> ReaderT SessionData IO ()
handleSettingsFrame _settings_list = do
let
enable_push = lookup NH2.SettingsEnablePush _settings_list
session_settings_mvar <- view sessionSettings
case enable_push of
Just 1 -> liftIO $ modifyMVar_ session_settings_mvar
$ \ old_settings -> return ( pushEnabled .~ True $ old_settings)
Just 0 -> liftIO $ modifyMVar_ session_settings_mvar
$ \ old_settings -> return ( pushEnabled .~ False $ old_settings)
Just _ -> closeConnectionBecauseIsInvalid NH2.ProtocolError
Nothing -> return ()
-- liftIO $ logit $ "Settings " `mappend` (pack . show $ _settings_list)
sendOutFrame
(NH2.EncodeInfo
(NH2.setAck NH2.defaultFlags)
0
Nothing
)
(NH2.SettingsFrame [])
sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()
sendOutFrame encode_info payload = do
session_output_mvar <- view sessionOutput
session_output <- liftIO $ takeMVar session_output_mvar
liftIO $ writeChan session_output $ Right (
encode_info,
payload,
-- Not sending effects in this frame, since it is not related...
error "sendOutFrameNotFor")
liftIO $ putMVar session_output_mvar session_output
-- TODO: This function, but using the headers editor, triggers
-- some renormalization of the header order. A good thing, if
-- I get that order well enough....
addExtraHeaders :: He.HeaderEditor -> ReaderT SessionData IO He.HeaderEditor
addExtraHeaders headers_editor = do
let
enriched_lens = (sessionsContext . sessionsConfig .sessionsEnrichedHeaders )
-- TODO: Figure out which is the best way to put this contact in the
-- source code
protocol_lens = He.headerLens "second-transfer-eh--used-protocol"
add_used_protocol <- view (enriched_lens . addUsedProtocol )
-- liftIO $ putStrLn $ "AAA" ++ (show add_used_protocol)
let
he1 = if add_used_protocol
then set protocol_lens (Just "HTTP/2") headers_editor
else headers_editor
if add_used_protocol
-- Nothing will be computed here if the headers are not modified.
then return he1
else return headers_editor
validateIncomingHeaders :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor)
validateIncomingHeaders headers_editor = do
-- Check that the headers block comes with all mandatory headers.
-- Right now I'm not checking that they come in the mandatory order though...
--
-- Notice that this function will transform a "host" header to an ":authority"
-- one.
let
h1 = He.replaceHostByAuthority headers_editor
-- Check that headers are lowercase
headers_are_lowercase = He.headersAreLowercaseAtHeaderEditor headers_editor
-- Check that we have mandatory headers
maybe_authority = h1 ^. (He.headerLens ":authority")
maybe_method = h1 ^. (He.headerLens ":method")
maybe_scheme = h1 ^. (He.headerLens ":scheme")
maybe_path = h1 ^. (He.headerLens ":path")
if
(isJust maybe_authority) &&
(isJust maybe_method) &&
(isJust maybe_scheme) &&
(isJust maybe_path )
then
return (Just h1)
else
return Nothing
-- Sends a GO_AWAY frame and raises an exception, effectively terminating the input
-- thread of the session.
closeConnectionBecauseIsInvalid :: NH2.ErrorCodeId -> ReaderT SessionData IO a
closeConnectionBecauseIsInvalid error_code = do
-- liftIO $ errorM "HTTP2.Session" "closeConnectionBecauseIsInvalid called!"
last_good_stream_mvar <- view lastGoodStream
last_good_stream <- liftIO $ takeMVar last_good_stream_mvar
session_output_mvar <- view sessionOutput
stream2workerthread <- view stream2WorkerThread
sendOutFrame
(NH2.EncodeInfo
NH2.defaultFlags
0
Nothing
)
(NH2.GoAwayFrame
last_good_stream
error_code
""
)
liftIO $ do
-- Close all active threads for this session
H.mapM_
( \(_stream_id, thread_id) ->
throwTo thread_id StreamCancelledException
)
stream2workerthread
-- Notify the framer that the session is closing, so
-- that it stops accepting frames from connected sources
-- (Streams?)
session_output <- takeMVar session_output_mvar
writeChan session_output $ Left CancelSession_SOC
putMVar session_output_mvar session_output
-- And unwind the input thread in the session, so that the
-- exception handler runs....
E.throw HTTP2ProtocolException
frameEndsStream :: InputFrame -> Bool
frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags
-- Executes its argument, unless receiving
-- headers, in which case the connection is closed.
unlessReceivingHeaders :: ReaderT SessionData IO a -> ReaderT SessionData IO a
unlessReceivingHeaders comp = do
receiving_headers_mvar <- view receivingHeaders
-- First check if we are receiving headers
maybe_recv_headers <- liftIO $ readMVar receiving_headers_mvar
if isJust maybe_recv_headers
then
-- So, this frame is highly illegal
closeConnectionBecauseIsInvalid NH2.ProtocolError
else
comp
createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism
createMechanismForStream stream_id = do
(chan, source) <- liftIO $ unfoldChannelAndSource
stream2postinputmechanism <- view stream2PostInputMechanism
let pim = PostInputMechanism (chan, source)
liftIO $ H.insert stream2postinputmechanism stream_id pim
return pim
-- TODO: Can be optimized by factoring out the mechanism lookup
-- TODO IMPORTANT: This is a good place to drop the postinputmechanism
-- for a stream, so that unprocessed data can be garbage-collected.
closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()
closePostDataSource stream_id = do
stream2postinputmechanism <- view stream2PostInputMechanism
pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id
case pim_maybe of
Just (PostInputMechanism (chan, _)) ->
liftIO $ putMVar chan Nothing
Nothing ->
-- TODO: This is a protocol error, handle it properly
error "Internal error/closePostDataSource"
streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()
streamWorkerSendData stream_id bytes = do
s2pim <- view stream2PostInputMechanism
pim_maybe <- liftIO $ H.lookup s2pim stream_id
case pim_maybe of
Just pim ->
sendBytesToPim pim bytes
Nothing ->
-- This is an internal error, the mechanism should be
-- created when the headers end (and if the headers
-- do not finish the stream)
error "Internal error"
sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()
sendBytesToPim (PostInputMechanism (chan, _)) bytes =
liftIO $ putMVar chan (Just bytes)
postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream
postDataSourceFromMechanism (PostInputMechanism (_, source)) = source
isSettingsAck :: NH2.FrameHeader -> Bool
isSettingsAck (NH2.FrameHeader _ flags _) =
NH2.testAck flags
isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool
isStreamCancelled stream_id = do
cancelled_streams_mvar <- view streamsCancelled_WTE
cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar
return $ NS.member stream_id cancelled_streams
sendPrimitive500Error :: IO TupledPrincipalStream
sendPrimitive500Error =
return (
[
(":status", "500")
],
[],
do
yield "Internal server error\n"
-- No footers
return []
)
-- Invokes the Coherent worker. Data is sent through pipes to
-- the output thread here in this session.
workerThread :: Request -> AwareWorker -> WorkerMonad ()
workerThread req aware_worker =
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
-- TODO: Handle exceptions here: what happens if the coherent worker
-- throws an exception signaling that the request is ill-formed
-- and should be dropped? That could happen in a couple of occassions,
-- but really most cases should be handled here in this file...
#if LOGIT_SWITCH_TIMINGS
liftIO . logit $ "worker-thread " `mappend` (pack . show $ stream_id)
#endif
-- (headers, _, data_and_conclussion)
principal_stream <-
liftIO $ E.catch
(
aware_worker req
)
(
const $ tupledPrincipalStreamToPrincipalStream <$> sendPrimitive500Error
:: HTTP500PrecursorException -> IO PrincipalStream
)
-- Pieces of the header
let
headers = principal_stream ^. headers_PS
data_and_conclusion = principal_stream ^. dataAndConclusion_PS
effects = principal_stream ^. effect_PS
pushed_streams = principal_stream ^. pushedStreams_PS
can_push = session_settings ^. pushEnabled
-- 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.
data_promises <- if can_push then do
forM pushed_streams $ \ pushed_stream_comp -> do
-- Initialize pushed stream
pushed_stream <- liftIO pushed_stream_comp
-- Send the request headers properly wrapped in a "Push Promise", do
-- it before sending the actual response headers of this stream
let
request_headers = pushed_stream ^. requestHeaders_Psh
response_headers = pushed_stream ^. responseHeaders_Psh
pushed_data_and_conclusion = pushed_stream ^. dataAndConclusion_Psh
child_stream_id <- liftIO $ modifyMVar next_push_stream_mvar
$ (\ x -> return (x+2,x) )
liftIO . writeChan headers_output . PushPromise_HM $
(stream_id, child_stream_id, request_headers, effects)
return (child_stream_id, response_headers, pushed_data_and_conclusion, effects)
else
return []
-- Now I send the headers, if that's possible at all
headers_sent <- liftIO newEmptyMVar
liftIO $ writeChan headers_output $ NormalResponse_HM (stream_id, headers_sent, headers, effects)
-- At this moment I should ask if the stream hasn't been cancelled by the browser before
-- commiting to the work of sending addtitional data... this is important for pushed
-- streams
is_stream_cancelled <- isStreamCancelled stream_id
unless is_stream_cancelled $ do
-- I have a beautiful source that I can de-construct...
-- TODO: Optionally pulling data out from a Conduit ....
-- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )
--
-- This threadlet should block here waiting for the headers to finish going
-- NOTE: Exceptions generated here inheriting from HTTP500PrecursorException
-- are let to bubble and managed in this thread fork point...
(_maybe_footers, _) <- runConduit $
transPipe liftIO data_and_conclusion
`fuseBothMaybe`
sendDataOfStream stream_id headers_sent effects
-- BIG TODO: Send the footers ... likely stream conclusion semantics
-- will need to be changed.
-- Now, time to fork threads for the pusher streams ....
forM_ data_promises
$ \ (child_stream_id, response_headers, pushed_data_and_conclusion, effects) -> do
environment <- ask
let
action = pusherThread
child_stream_id
response_headers
pushed_data_and_conclusion
effects
-- And let the action run in its own thread
liftIO . forkIO $ runReaderT action environment
return ()
-- Invokes the Coherent worker. Data is sent through pipes to
-- the output thread here in this session.
pusherThread :: GlobalStreamId -> Headers -> DataAndConclusion -> Effect -> WorkerMonad ()
pusherThread child_stream_id response_headers pushed_data_and_conclusion effects =
do
headers_output <- view headersOutput
session_settings_mvar <- view sessionSettings_WTE
session_settings <- liftIO $ readMVar session_settings_mvar
-- TODO: Handle exceptions here: what happens if the coherent worker
-- throws an exception signaling that the request is ill-formed
-- and should be dropped? That could happen in a couple of occassions,
-- but really most cases should be handled here in this file...
-- (headers, _, data_and_conclussion)
-- Now I send the headers, if that's possible at all
headers_sent <- liftIO newEmptyMVar
liftIO . writeChan headers_output
$ NormalResponse_HM (child_stream_id, headers_sent, response_headers, effects)
-- At this moment I should ask if the stream hasn't been cancelled by the browser before
-- commiting to the work of sending addtitional data... this is important for pushed
-- streams
is_stream_cancelled <- isStreamCancelled child_stream_id
unless is_stream_cancelled $ do
-- I have a beautiful source that I can de-construct...
-- TODO: Optionally pulling data out from a Conduit ....
-- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )
--
-- This threadlet should block here waiting for the headers to finish going
-- NOTE: Exceptions generated here inheriting from HTTP500PrecursorException
-- are let to bubble and managed in this thread fork point...
(_maybe_footers, _) <- runConduit $
transPipe liftIO pushed_data_and_conclusion
`fuseBothMaybe`
sendDataOfStream child_stream_id headers_sent effects
-- BIG TODO: Send the footers ... likely stream conclusion semantics
-- will need to be changed.
return ()
-- v-- comp. monad.
sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Effect -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()
sendDataOfStream stream_id headers_sent effect = do
data_output <- view dataOutput
-- Wait for all headers sent
liftIO $ takeMVar headers_sent
consumer data_output
where
--delay = effect ^. middlePauseForDelivery_Ef
consumer data_output = do
maybe_bytes <- await
case maybe_bytes of
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
-- Returns if the frame is the first in the stream
appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO Bool
appendHeaderFragmentBlock global_stream_id bytes = do
ht <- view stream2HeaderBlockFragment
maybe_old_block <- liftIO $ H.lookup ht global_stream_id
(new_block, new_stream) <- case maybe_old_block of
Nothing -> do
-- TODO: Make the commented message below more informative
return $ (Bu.byteString bytes, True)
Just something ->
return $ (something `mappend` (Bu.byteString bytes), False)
liftIO $ H.insert ht global_stream_id new_block
return new_stream
getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString
getHeaderBytes global_stream_id = do
ht <- view stream2HeaderBlockFragment
Just bytes <- liftIO $ H.lookup ht global_stream_id
return $ Bl.toStrict $ Bu.toLazyByteString bytes
isAboutHeaders :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)
isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) )
= Just (stream_id, block_fragment)
isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )
= Just (stream_id, block_fragment)
isAboutHeaders _
= Nothing
getHeadersPriority :: InputFrame -> Maybe NH2.Priority
getHeadersPriority (NH2.Frame _ ( NH2.HeadersFrame prio _ ) ) = prio
getHeadersPriority _ = Nothing
frameEndsHeaders :: InputFrame -> Bool
frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags
streamIdFromFrame :: InputFrame -> GlobalStreamId
streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = stream_id
-- TODO: Have different size for the headers..... just now going with a default size of 16 k...
-- TODO: Find a way to kill this thread....
headersOutputThread :: Chan HeaderOutputMessage -- (GlobalStreamId, MVar HeadersSent, Headers, Effect)
-> MVar SessionOutputChannelAbstraction
-> ReaderT SessionData IO ()
headersOutputThread input_chan session_output_mvar = forever $ do
use_chunk_length <- view $ sessionsContext . sessionsConfig . dataFrameSize
header_output_request <- liftIO $ readChan input_chan
case header_output_request of
NormalResponse_HM (stream_id, headers_ready_mvar, headers, effect) -> do
-- First encode the headers using the table
encode_dyn_table_mvar <- view toEncodeHeaders
encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar
(new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers
liftIO $ putMVar encode_dyn_table_mvar new_dyn_table
-- Now split the bytestring in chunks of the needed size....
-- Note that the only way we can
let
bs_chunks = bytestringChunk use_chunk_length data_to_send
-- And send the chunks through while locking the output place....
liftIO $ bs_chunks `deepseq` E.bracket
(takeMVar session_output_mvar)
(putMVar session_output_mvar )
(\ session_output -> do
writeIndividualHeaderFrames session_output stream_id bs_chunks True effect
-- And say that the headers for this thread are out
-- INSTRUMENTATION( debugM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )
putMVar headers_ready_mvar HeadersSent
)
PushPromise_HM (parent_stream_id, child_stream_id, promise_headers, effect) -> do
encode_dyn_table_mvar <- view toEncodeHeaders
encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar
(new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table promise_headers
liftIO $ putMVar encode_dyn_table_mvar new_dyn_table
-- Now split the bytestring in chunks of the needed size....
bs_chunks <- return $! bytestringChunk use_chunk_length data_to_send
-- And send the chunks through while locking the output place....
liftIO $ E.bracket
(takeMVar session_output_mvar)
(putMVar session_output_mvar )
(\ session_output -> do
writePushPromiseFrames session_output parent_stream_id child_stream_id bs_chunks True effect
-- No notification here... but since everyhing headers is serialized here and the next
-- piece of information corresponding to the child stream is a headers sequence, we shouldn't
-- need any other tricks here
)
where
writeIndividualHeaderFrames ::
SessionOutputChannelAbstraction
-> GlobalStreamId
-> [B.ByteString]
-> Bool
-> Effect
-> IO ()
writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first effect =
writeChan session_output $ Right ( NH2.EncodeInfo {
NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags
,NH2.encodeStreamId = stream_id
,NH2.encodePadding = Nothing },
(if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment),
effect
)
writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first effect = do
writeChan session_output $ Right ( NH2.EncodeInfo {
NH2.encodeFlags = NH2.defaultFlags
,NH2.encodeStreamId = stream_id
,NH2.encodePadding = Nothing },
(if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment),
effect
)
writeIndividualHeaderFrames session_output stream_id xs False effect
writePushPromiseFrames ::
SessionOutputChannelAbstraction
-> GlobalStreamId
-> GlobalStreamId
-> [B.ByteString]
-> Bool
-> Effect
-> IO ()
writePushPromiseFrames session_output parent_stream_id child_stream_id (last_fragment:[]) is_first effect =
writeChan session_output $ Right ( NH2.EncodeInfo {
NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags
,NH2.encodeStreamId = parent_stream_id
,NH2.encodePadding = Nothing },
(if is_first
then NH2.PushPromiseFrame child_stream_id last_fragment
else NH2.ContinuationFrame last_fragment
),
effect
)
writePushPromiseFrames session_output parent_stream_id child_stream_id (fragment:xs) is_first effect = do
writeChan session_output $ Right ( NH2.EncodeInfo {
NH2.encodeFlags = NH2.defaultFlags
,NH2.encodeStreamId = parent_stream_id
,NH2.encodePadding = Nothing },
(if is_first
then NH2.PushPromiseFrame child_stream_id fragment
else NH2.ContinuationFrame fragment
),
effect
)
writePushPromiseFrames session_output parent_stream_id child_stream_id xs False effect
bytestringChunk :: Int -> B.ByteString -> [B.ByteString]
bytestringChunk len s | (B.length s) < len = [ s ]
bytestringChunk len s = h:(bytestringChunk len xs)
where
(h, xs) = B.splitAt len s
-- This thread is for the entire session, not for each stream. The thread should
-- die while waiting for input in a garbage-collected mvar.
-- TODO: This function does non-optimal chunking for the case where responses are
-- actually streamed.... in those cases we need to keep state for frames in
-- some other format....
-- TODO: Right now, we are transmitting an empty last frame with the end-of-stream
-- flag set. I'm afraid that the only
-- way to avoid that is by holding a frame or by augmenting the end-user interface
-- so that the user can signal which one is the last frame. The first approach
-- restricts responsiviness, the second one clutters things.
dataOutputThread :: Int
-> MVar DataOutputToConveyor
-> MVar SessionOutputChannelAbstraction
-> IO ()
dataOutputThread use_chunk_length input_chan session_output_mvar = forever $ do
(stream_id, maybe_contents, effect) <- takeMVar input_chan
case maybe_contents of
Nothing -> do
liftIO $ do
withLockedSessionOutput
(\ session_output -> do
-- Write an empty data-frame with the right flags
writeChan session_output $ Right ( NH2.EncodeInfo {
NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags
,NH2.encodeStreamId = stream_id
,NH2.encodePadding = Nothing },
NH2.DataFrame "",
effect
)
-- And then write an-end-of-stream command
writeChan session_output $ Left . FinishStream_SOC $ stream_id
)
Just contents -> do
-- And now just simply output it...
let bs_chunks = bytestringChunk use_chunk_length contents
-- And send the chunks through while locking the output place....
bs_chunks `deepseq` writeContinuations bs_chunks stream_id effect
return ()
where
withLockedSessionOutput = E.bracket
(takeMVar session_output_mvar)
(putMVar session_output_mvar) -- <-- There is an implicit argument there!!
writeContinuations :: [B.ByteString] -> GlobalStreamId -> Effect -> IO ()
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 )
)
)
fragments