git.iko.soy/typeable/wai
on master help sign in
Merging branch for "HTTP/2 performance regression" (#470)
commit1da6177909873c5de1e23e46c32772e4b5744d99 authorKazu Yamamoto <kazu@iij.ad.jp> dateWed 16 Dec 2015, 14:27 (11 years ago) parent565a48929f4ddb
19 files changed+584−1181
compare with master
unifiedsplit
wai/Network/Wai/HTTP2.hs−272 wai/Network/Wai/Internal.hs−97 wai/wai.cabal−2 warp-tls/Network/Wai/Handler/WarpTLS.hs+7−29 warp/Network/Wai/Handler/Warp.hs−9 warp/Network/Wai/Handler/Warp/File.hs+1 warp/Network/Wai/Handler/Warp/HTTP2.hs+6−9 warp/Network/Wai/Handler/Warp/HTTP2/HPACK.hs+8−19 warp/Network/Wai/Handler/Warp/HTTP2/Manager.hs+31−31 warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+15−25 warp/Network/Wai/Handler/Warp/HTTP2/Request.hs+46−28 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+238−228 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+75−141 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+126−173 warp/Network/Wai/Handler/Warp/Internal.hs−11 warp/Network/Wai/Handler/Warp/RequestHeader.hs+1−2 warp/Network/Wai/Handler/Warp/Response.hs+1 warp/Network/Wai/Handler/Warp/Run.hs+28−104 warp/test/RequestSpec.hs+1−1
wai/Network/Wai/HTTP2.hsdeleted−272
@@ -1,272 +0,0 @@
1−{-# LANGUAGE OverloadedStrings, RankNTypes #-}
2−{-# LANGUAGE CPP #-}
3−{-# LANGUAGE DeriveDataTypeable #-}
4−
5−-- | An HTTP\/2-aware variant of the 'Network.Wai.Application' type. Compared
6−-- to the original, this exposes the new functionality of server push and
7−-- trailers, allows stream fragments to be sent in the form of file ranges, and
8−-- allows the stream body to produce a value to be used in constructing the
9−-- trailers. Existing @Applications@ can be faithfully upgraded to HTTP\/2
10−-- with 'promoteApplication' or served transparently over both protocols with
11−-- the normal Warp 'Network.Wai.Handler.Warp.run' family of functions.
12−--
13−-- An 'HTTP2Application' takes a 'Request' and a 'PushFunc' and produces a
14−-- 'Responder' that will push any associated resources and send the response
15−-- body. The response is always a stream of 'Builder's and file chunks.
16−-- Equivalents of the 'Network.Wai.responseBuilder' family of functions are
17−-- provided for creating 'Responder's conveniently.
18−--
19−-- Pushed streams are handled by an IO action that triggers a server push. It
20−-- returns @True@ if the @PUSH_PROMISE@ frame was sent, @False@ if not. Note
21−-- this means it will still return @True@ if the client reset or ignored the
22−-- stream. This gives handlers the freedom to implement their own heuristics
23−-- for whether to actually push a resource, while also allowing middleware and
24−-- frameworks to trigger server pushes automatically.
25−
26−module Network.Wai.HTTP2
27− (
28− -- * Applications
29− HTTP2Application
30− -- * Responder
31− , Responder(..)
32− , RespondFunc
33− , Body
34− , Chunk(..)
35− , Trailers
36− -- * Server push
37− , PushFunc
38− , PushPromise(..)
39− , promiseHeaders
40− -- * Conveniences
41− , promoteApplication
42− -- ** Responders
43− , respond
44− , respondCont
45− , respondIO
46− , respondFile
47− , respondFilePart
48− , respondNotFound
49− , respondWith
50− -- ** Stream Bodies
51− , streamFilePart
52− , streamBuilder
53− , streamSimple
54− ) where
55−
56−import Blaze.ByteString.Builder (Builder)
57−import Blaze.ByteString.Builder.ByteString (fromByteString)
58−import Control.Exception (Exception, throwIO)
59−import Control.Monad.Trans.Cont (ContT(..))
60−import Data.ByteString (ByteString)
61−#if __GLASGOW_HASKELL__ < 709
62−import Data.Functor ((<$>))
63−#endif
64−import Data.IORef (newIORef, readIORef, writeIORef)
65−#if __GLASGOW_HASKELL__ < 709
66−import Data.Monoid (mempty)
67−#endif
68−import Data.Typeable (Typeable)
69−import qualified Network.HTTP.Types as H
70−
71−import Network.Wai (Application)
72−import Network.Wai.Internal
73− ( FilePart(..)
74− , Request(requestHeaders)
75− , Response(..)
76− , ResponseReceived(..)
77− , StreamingBody
78− , adjustForFilePart
79− , chooseFilePart
80− , tryGetFileSize
81− )
82−
83−-- | Headers sent after the end of a data stream, as defined by section 4.1.2 of
84−-- the HTTP\/1.1 spec (RFC 7230), and section 8.1 of the HTTP\/2 spec.
85−type Trailers = [H.Header]
86−
87−-- | The synthesized request and headers of a pushed stream.
88−data PushPromise = PushPromise
89− { promisedMethod :: H.Method
90− , promisedPath :: ByteString
91− , promisedAuthority :: ByteString
92− , promisedScheme :: ByteString
93− , promisedHeader :: H.RequestHeaders
94− }
95−
96−-- | The HTTP\/2-aware equivalent of 'Network.Wai.Application'.
97−type HTTP2Application = Request -> PushFunc -> Responder
98−
99−-- | Part of a streaming response -- either a 'Builder' or a range of a file.
100−data Chunk = FileChunk FilePath FilePart | BuilderChunk Builder
101−
102−-- | The streaming body of a response. Equivalent to
103−-- 'Network.Wai.StreamingBody' except that it can also write file ranges and
104−-- return the stream's trailers.
105−type Body = (Chunk -> IO ()) -> IO () -> IO Trailers
106−
107−-- | Given to 'Responders'; provide a status, headers, and a stream body, and
108−-- we'll give you a token proving you called the 'RespondFunc'.
109−type RespondFunc s = H.Status -> H.ResponseHeaders -> Body -> IO s
110−
111−-- | The result of an 'HTTP2Application'; or, alternately, an application
112−-- that's independent of the request. This is a continuation-passing style
113−-- function that first provides a response by calling the given respond
114−-- function, then returns the request's 'Trailers'.
115−--
116−-- The respond function is similar to the one in 'Network.Wai.Application', but
117−-- it only takes a streaming body, the status and headers are curried, and it
118−-- also produces trailers for the stream.
119−newtype Responder = Responder
120− { runResponder :: forall s. RespondFunc s -> IO s }
121−
122−-- | A function given to an 'HTTP2Application' to initiate a server-pushed
123−-- stream. Its argument is the same as the result of an 'HTTP2Application', so
124−-- you can either implement the response inline, or call your own application
125−-- to create the response.
126−--
127−-- The result is 'True' if the @PUSH_PROMISE@ frame will be sent, or 'False' if
128−-- it will not. This can happen if server push is disabled, the concurrency
129−-- limit of server-initiated streams is reached, or the associated stream has
130−-- already been closed.
131−--
132−-- This function shall ensure that stream data provided after it returns will
133−-- be sent after the @PUSH_PROMISE@ frame, so that servers can implement the
134−-- requirement that any pushed stream for a resource be initiated before
135−-- sending DATA frames that reference it.
136−type PushFunc = PushPromise -> Responder -> IO Bool
137−
138−-- | Create the 'H.RequestHeaders' corresponding to the given 'PushPromise'.
139−--
140−-- This is primarily useful for WAI handlers like Warp, and application
141−-- implementers are unlikely to use it directly.
142−promiseHeaders :: PushPromise -> H.RequestHeaders
143−promiseHeaders p =
144− [ (":method", promisedMethod p)
145− , (":path", promisedPath p)
146− , (":authority", promisedAuthority p)
147− , (":scheme", promisedScheme p)
148− ] ++ promisedHeader p
149−
150−-- | Create a response body consisting of a single range of a file. Does not
151−-- set Content-Length or Content-Range headers. For that, use
152−-- 'respondFilePart' or 'respondFile'.
153−streamFilePart :: FilePath -> FilePart -> Body
154−streamFilePart path part write _ = write (FileChunk path part) >> return []
155−
156−-- | Respond with a single range of a file, adding the Accept-Ranges,
157−-- Content-Length and Content-Range headers and changing the status to 206 as
158−-- appropriate.
159−--
160−-- If you want the range to be inferred automatically from the Range header,
161−-- use 'respondFile' instead. On the other hand, if you want to avoid the
162−-- automatic header and status adjustments, use 'respond' and 'streamFilePart'
163−-- directly.
164−respondFilePart :: H.Status -> H.ResponseHeaders -> FilePath -> FilePart -> Responder
165−respondFilePart s h path part = Responder $ \k -> do
166− let (s', h') = adjustForFilePart s h part
167− k s' h' $ streamFilePart path part
168−
169−-- | Serve the requested range of the specified file (based on the Range
170−-- header), using the given 'H.Status' and 'H.ResponseHeaders' as a base. If
171−-- the file is not accessible, the status will be replaced with 404 and a
172−-- default not-found message will be served. If a partial file is requested,
173−-- the status will be replaced with 206 and the Content-Range header will be
174−-- added. The Content-Length header will always be added.
175−respondFile :: H.Status -> H.ResponseHeaders -> FilePath -> H.RequestHeaders -> Responder
176−respondFile s h path reqHdrs = Responder $ \k -> do
177− fileSize <- tryGetFileSize path
178− case fileSize of
179− Left _ -> runResponder (respondNotFound h) k
180− Right size -> runResponder (respondFileExists s h path size reqHdrs) k
181−
182−-- As 'respondFile', but with prior knowledge of the file's existence and size.
183−respondFileExists :: H.Status -> H.ResponseHeaders -> FilePath -> Integer -> H.RequestHeaders -> Responder
184−respondFileExists s h path size reqHdrs =
185− respondFilePart s h path $ chooseFilePart size $ lookup H.hRange reqHdrs
186−
187−-- | Respond with a minimal 404 page with the given headers.
188−respondNotFound :: H.ResponseHeaders -> Responder
189−respondNotFound h = Responder $ \k -> k H.notFound404 h' $
190− streamBuilder $ fromByteString "File not found."
191− where
192− contentType = (H.hContentType, "text/plain; charset=utf-8")
193− h' = contentType:filter ((/=H.hContentType) . fst) h
194−
195−-- | Construct a 'Responder' that will just call the 'RespondFunc' with the
196−-- given arguments.
197−respond :: H.Status -> H.ResponseHeaders -> Body -> Responder
198−respond s h b = Responder $ \k -> k s h b
199−
200−-- | Fold the given bracketing action into a 'Responder'. Note the first
201−-- argument is isomorphic to @Codensity IO a@ or @forall s. ContT s IO a@, and
202−-- is the type of a partially-applied 'Control.Exception.bracket' or
203−-- @with@-style function.
204−--
205−-- > respondWith (bracket acquire release) $
206−-- > \x -> respondNotFound [("x", show x)]
207−--
208−-- is equivalent to
209−--
210−-- > Responder $ \k -> bracket acquire release $
211−-- > \x -> runResponder (respondNotFound [("x", show x)] k
212−--
213−-- This is morally equivalent to ('>>=') on 'Codensity' 'IO'.
214−respondWith :: (forall s. (a -> IO s) -> IO s) -> (a -> Responder) -> Responder
215−respondWith with f = respondCont $ f <$> ContT with
216−
217−-- | Fold the 'ContT' into the contained 'Responder'.
218−respondCont :: (forall r. ContT r IO Responder) -> Responder
219−respondCont cont = Responder $ \k -> runContT cont $ \r -> runResponder r k
220−
221−-- | Fold the 'IO' into the contained 'Responder'.
222−respondIO :: IO Responder -> Responder
223−respondIO io = Responder $ \k -> io >>= \r -> runResponder r k
224−
225−-- | Create a response body consisting of a single builder.
226−streamBuilder :: Builder -> Body
227−streamBuilder builder write _ = write (BuilderChunk builder) >> return []
228−
229−-- | Create a response body of a stream of 'Builder's.
230−streamSimple :: StreamingBody -> Body
231−streamSimple body write flush = body (write . BuilderChunk) flush >> return []
232−
233−-- | Use a normal WAI 'Response' to send the response. Useful if you're
234−-- sharing code between HTTP\/2 applications and HTTP\/1 applications.
235−--
236−-- The 'Request' is used to determine the right file range to serve for
237−-- 'ResponseFile'.
238−promoteResponse :: Request -> Response -> Responder
239−promoteResponse req response = case response of
240− (ResponseBuilder s h b) ->
241− Responder $ \k -> k s h (streamBuilder b)
242− (ResponseStream s h body) ->
243− Responder $ \k -> k s h (streamSimple body)
244− (ResponseRaw _ fallback) -> promoteResponse req fallback
245− (ResponseFile s h path mpart) -> maybe
246− (respondFile s h path $ requestHeaders req)
247− (respondFilePart s h path)
248− mpart
249−
250−-- | An 'Network.Wai.Application' we tried to promote neither called its
251−-- respond action nor raised; this is only possible if it imported the
252−-- 'ResponseReceived' constructor and used it to lie about having called the
253−-- action.
254−data RespondNeverCalled = RespondNeverCalled deriving (Show, Typeable)
255−
256−instance Exception RespondNeverCalled
257−
258−-- | Promote a normal WAI 'Application' to an 'HTTP2Application' by ignoring
259−-- the HTTP/2-specific features.
260−promoteApplication :: Application -> HTTP2Application
261−promoteApplication app req _ = Responder $ \k -> do
262− -- In HTTP2Applications, the Responder is required to ferry a value of
263− -- arbitrary type from the RespondFunc back to the caller of the
264− -- application, but in Application the type is fixed to ResponseReceived.
265− -- To add this extra power to an Application, we have to squirrel it away
266− -- in an IORef as a hack.
267− ref <- newIORef Nothing
268− let k' r = do
269− writeIORef ref . Just =<< runResponder (promoteResponse req r) k
270− return ResponseReceived
271− ResponseReceived <- app req k'
272− readIORef ref >>= maybe (throwIO RespondNeverCalled) return
wai/Network/Wai/Internal.hs−97
@@ -1,7 +1,6 @@
11{-# OPTIONS_HADDOCK not-home #-}
22{-# LANGUAGE DeriveDataTypeable #-}
33{-# LANGUAGE CPP #-}
4−{-# LANGUAGE OverloadedStrings #-}
54{-# LANGUAGE RankNTypes #-}
65{-# LANGUAGE RecordWildCards #-}
76-- | Internal constructors and helper functions. Note that no guarantees are
@@ -9,25 +8,17 @@
98module Network.Wai.Internal where
109
1110import Blaze.ByteString.Builder (Builder)
12−import Control.Exception (IOException, try)
1311import qualified Data.ByteString as B hiding (pack)
14−import qualified Data.ByteString.Builder as B
15−import qualified Data.ByteString.Char8 as B (pack, readInteger)
16−import qualified Data.ByteString.Lazy as L
1712#if __GLASGOW_HASKELL__ < 709
1813import Data.Functor ((<$>))
1914#endif
20−import Data.Maybe (listToMaybe)
2115import Data.Text (Text)
2216import Data.Typeable (Typeable)
2317import Data.Vault.Lazy (Vault)
2418import Data.Word (Word64)
2519import qualified Network.HTTP.Types as H
26−import qualified Network.HTTP.Types.Header as HH
2720import Network.Socket (SockAddr)
28−import Numeric (showInt)
2921import Data.List (intercalate)
30−import qualified System.PosixCompat.Files as P
3122
3223-- | Information on the request sent by the client. This abstracts away the
3324-- details of the underlying implementation.
@@ -157,91 +148,3 @@ data FilePart = FilePart
157148-- Since 3.0.0
158149data ResponseReceived = ResponseReceived
159150 deriving Typeable
160−
161−-- | Look up the size of a file in 'Right' or the 'IOException' in 'Left'.
162−tryGetFileSize :: FilePath -> IO (Either IOException Integer)
163−tryGetFileSize path =
164− fmap (fromIntegral . P.fileSize) <$> try (P.getFileStatus path)
165−
166−-- | \"Content-Range\".
167−hContentRange :: H.HeaderName
168−hContentRange = "Content-Range"
169−
170−-- | \"Accept-Ranges\".
171−hAcceptRanges :: H.HeaderName
172−hAcceptRanges = "Accept-Ranges"
173−
174−-- | @contentRangeHeader beg end total@ constructs a Content-Range 'H.Header'
175−-- for the range specified.
176−contentRangeHeader :: Integer -> Integer -> Integer -> H.Header
177−contentRangeHeader beg end total = (hContentRange, range)
178− where
179− range = B.pack
180− -- building with ShowS
181− $ 'b' : 'y': 't' : 'e' : 's' : ' '
182− : (if beg > end then ('*':) else
183− showInt beg
184− . ('-' :)
185− . showInt end)
186− ( '/'
187− : showInt total "")
188−
189−-- | Given the full size of a file and optionally a Range header value,
190−-- determine the range to serve by parsing the range header and obeying it, or
191−-- serving the whole file if it's absent or malformed.
192−chooseFilePart :: Integer -> Maybe B.ByteString -> FilePart
193−chooseFilePart size Nothing = FilePart 0 size size
194−chooseFilePart size (Just range) = case parseByteRanges range >>= listToMaybe of
195− -- Range is broken
196− Nothing -> FilePart 0 size size
197− Just hrange -> checkRange hrange
198− where
199− checkRange (H.ByteRangeFrom beg) = fromRange beg (size - 1)
200− checkRange (H.ByteRangeFromTo beg end) = fromRange beg (min (size - 1) end)
201− checkRange (H.ByteRangeSuffix count) = fromRange (max 0 (size - count)) (size - 1)
202−
203− fromRange beg end = FilePart beg (end - beg + 1) size
204−
205−-- | Adjust the given 'H.Status' and 'H.ResponseHeaders' based on the given
206−-- 'FilePart'. This means replacing the status with 206 if the response is
207−-- partial, and adding the Content-Length and Accept-Ranges (always) and
208−-- Content-Range (if appropriate) headers.
209−adjustForFilePart :: H.Status -> H.ResponseHeaders -> FilePart -> (H.Status, H.ResponseHeaders)
210−adjustForFilePart s h part = (s', h'')
211− where
212− off = filePartOffset part
213− len = filePartByteCount part
214− size = filePartFileSize part
215−
216− contentRange = contentRangeHeader off (off + len - 1) size
217− lengthBS = L.toStrict $ B.toLazyByteString $ B.integerDec len
218− s' = if filePartByteCount part /= size then H.partialContent206 else s
219− h' = (H.hContentLength, lengthBS):(hAcceptRanges, "bytes"):h
220− h'' = (if len == size then id else (contentRange:)) h'
221−
222−-- | Parse the value of a Range header into a 'HH.ByteRanges'.
223−parseByteRanges :: B.ByteString -> Maybe HH.ByteRanges
224−parseByteRanges bs1 = do
225− bs2 <- stripPrefix "bytes=" bs1
226− (r, bs3) <- range bs2
227− ranges (r:) bs3
228− where
229− range bs2 = do
230− (i, bs3) <- B.readInteger bs2
231− if i < 0 -- has prefix "-" ("-0" is not valid, but here treated as "0-")
232− then Just (HH.ByteRangeSuffix (negate i), bs3)
233− else do
234− bs4 <- stripPrefix "-" bs3
235− case B.readInteger bs4 of
236− Just (j, bs5) | j >= i -> Just (HH.ByteRangeFromTo i j, bs5)
237− _ -> Just (HH.ByteRangeFrom i, bs4)
238− ranges front bs3
239− | B.null bs3 = Just (front [])
240− | otherwise = do
241− bs4 <- stripPrefix "," bs3
242− (r, bs5) <- range bs4
243− ranges (front . (r:)) bs5
244−
245− stripPrefix x y
246− | x `B.isPrefixOf` y = Just (B.drop (B.length x) y)
247− | otherwise = Nothing
wai/wai.cabal−2
@@ -27,10 +27,8 @@ Library
2727 , http-types >= 0.7
2828 , text >= 0.7
2929 , transformers >= 0.0
30− , unix-compat >= 0.2
3130 , vault >= 0.3 && < 0.4
3231 Exposed-modules: Network.Wai
33− Network.Wai.HTTP2
3432 Network.Wai.Internal
3533 ghc-options: -Wall
3634
warp-tls/Network/Wai/Handler/WarpTLS.hs+7−29
@@ -35,8 +35,6 @@ module Network.Wai.Handler.WarpTLS (
3535 -- * Runner
3636 , runTLS
3737 , runTLSSocket
38− , runHTTP2TLS
39− , runHTTP2TLSSocket
4038 -- * Exception
4139 , WarpTLSException (..)
4240 ) where
@@ -60,7 +58,6 @@ import Network.Socket.ByteString (sendAll)
6058import qualified Network.TLS as TLS
6159import qualified Network.TLS.Extra as TLSExtra
6260import Network.Wai (Application)
63−import Network.Wai.HTTP2 (HTTP2Application)
6461import Network.Wai.Handler.Warp
6562import Network.Wai.Handler.Warp.Internal
6663import System.IO.Error (isEOFError)
@@ -221,39 +218,20 @@ tlsSettingsChainMemory cert chainCerts key = defaultTlsSettings
221218
222219----------------------------------------------------------------
223220
224−-- Composition over two arguments at once; used for runHTTP2\*.
225−(.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d
226−f .: g = curry $ f . uncurry g
227−
228221-- | Running 'Application' with 'TLSSettings' and 'Settings'.
229222runTLS :: TLSSettings -> Settings -> Application -> IO ()
230−runTLS tset set = runServeTLS tset set . serveDefault
231−
232−-- | Run an HTTP\/2-aware server with 'TLSSettings' and 'Settings'.
233−runHTTP2TLS :: TLSSettings -> Settings -> HTTP2Application -> Application -> IO ()
234−runHTTP2TLS tset set = runServeTLS tset set .: serveHTTP2
235−
236−runServeTLS :: TLSSettings -> Settings -> ServeConnection -> IO ()
237−runServeTLS tset set serve = withSocketsDo $
223+runTLS tset set app = withSocketsDo $
238224 bracket
239225 (bindPortTCP (getPort set) (getHost set))
240226 sClose
241− (\sock -> runServeTLSSocket tset set sock serve)
227+ (\sock -> runTLSSocket tset set sock app)
242228
243229----------------------------------------------------------------
244230
245231-- | Running 'Application' with 'TLSSettings' and 'Settings' using
246232-- specified 'Socket'.
247233runTLSSocket :: TLSSettings -> Settings -> Socket -> Application -> IO ()
248−runTLSSocket tset set sock = runServeTLSSocket tset set sock . serveDefault
249−
250−-- | Run an HTTP\/2-aware server with 'TLSSettings' and 'Settings' using
251−-- specified 'Socket'.
252−runHTTP2TLSSocket :: TLSSettings -> Settings -> Socket -> HTTP2Application -> Application -> IO ()
253−runHTTP2TLSSocket tset set sock = runServeTLSSocket tset set sock .: serveHTTP2
254−
255−runServeTLSSocket :: TLSSettings -> Settings -> Socket -> ServeConnection -> IO ()
256−runServeTLSSocket tlsset@TLSSettings{..} set sock serve = do
234+runTLSSocket tlsset@TLSSettings{..} set sock app = do
257235 credential <- case (certMemory, keyMemory) of
258236 (Nothing, Nothing) ->
259237 either error id <$>
@@ -263,11 +241,11 @@ runServeTLSSocket tlsset@TLSSettings{..} set sock serve = do
263241 key <- maybe (S.readFile keyFile) return mkey
264242 either error return $
265243 TLS.credentialLoadX509ChainFromMemory cert chainCertsMemory key
266− runServeTLSSocket' tlsset set credential sock serve
244+ runTLSSocket' tlsset set credential sock app
267245
268−runServeTLSSocket' :: TLSSettings -> Settings -> TLS.Credential -> Socket -> ServeConnection -> IO ()
269−runServeTLSSocket' tlsset@TLSSettings{..} set credential sock serve =
270− runServeSettingsConnectionMakerSecure set get serve
246+runTLSSocket' :: TLSSettings -> Settings -> TLS.Credential -> Socket -> Application -> IO ()
247+runTLSSocket' tlsset@TLSSettings{..} set credential sock app =
248+ runSettingsConnectionMakerSecure set get app
271249 where
272250 get = getter tlsset sock params
273251 params = TLS.ServerParams {
warp/Network/Wai/Handler/Warp.hs−9
@@ -44,15 +44,6 @@ module Network.Wai.Handler.Warp (
4444 , runEnv
4545 , runSettings
4646 , runSettingsSocket
47− -- * Run an HTTP\/2-aware server
48− -- | Each of these takes an HTTP\/2-aware application as well as a backup
49− -- 'Application' to be used for HTTP\/1.1 and HTTP\/1 connections. These
50− -- are only needed if your application needs access to HTTP\/2-specific
51− -- features such as trailers or pushed streams.
52− , runHTTP2
53− , runHTTP2Env
54− , runHTTP2Settings
55− , runHTTP2SettingsSocket
5647 -- * Settings
5748 , Settings
5849 , defaultSettings
warp/Network/Wai/Handler/Warp/File.hs+1
@@ -6,6 +6,7 @@ module Network.Wai.Handler.Warp.File (
66 RspFileInfo(..)
77 , conditionalRequest
88 , addContentHeadersForFilePart
9+ , parseByteRanges
910 ) where
1011
1112import Control.Applicative ((<|>))
warp/Network/Wai/Handler/Warp/HTTP2.hs+6−9
@@ -7,11 +7,9 @@ import Control.Concurrent (forkIO, killThread)
77import qualified Control.Exception as E
88import Control.Monad (when, unless, replicateM_)
99import Data.ByteString (ByteString)
10−
1110import Network.HTTP2
1211import Network.Socket (SockAddr)
13−
14−import Network.Wai.HTTP2 (HTTP2Application)
12+import Network.Wai
1513import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
1614import Network.Wai.Handler.Warp.HTTP2.Manager
1715import Network.Wai.Handler.Warp.HTTP2.Receiver
@@ -24,16 +22,16 @@ import Network.Wai.Handler.Warp.Types
2422
2523----------------------------------------------------------------
2624
27−http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> HTTP2Application -> IO ()
25+http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> Application -> IO ()
2826http2 conn ii addr transport settings readN app = do
2927 checkTLS
3028 ok <- checkPreface
3129 when ok $ do
3230 ctx <- newContext
33− -- Workers & Manager
34− mgr <- start
35− let responder = response ctx mgr
36− action = worker ctx settings tm app responder
31+ -- Workers, worker manager and timer manager
32+ mgr <- start settings
33+ let responder = response ii settings ctx mgr
34+ action = worker ctx settings app responder
3735 setAction mgr action
3836 -- fixme: hard coding: 10
3937 replicateM_ 10 $ spawnAction mgr
@@ -48,7 +46,6 @@ http2 conn ii addr transport settings readN app = do
4846 stop mgr
4947 killThread tid
5048 where
51− tm = timeoutManager ii
5249 checkTLS = case transport of
5350 TCP -> return () -- direct
5451 tls -> unless (tls12orLater tls) $ goaway conn InadequateSecurity "Weak TLS"
warp/Network/Wai/Handler/Warp/HTTP2/HPACK.hs+8−19
@@ -13,6 +13,7 @@ import Data.IORef (readIORef, writeIORef)
1313import Network.HPACK
1414import qualified Network.HTTP.Types as H
1515import Network.HTTP2
16+import Network.Wai
1617import Network.Wai.Handler.Warp.HTTP2.Types
1718import Network.Wai.Handler.Warp.Header
1819import Network.Wai.Handler.Warp.Response
@@ -24,14 +25,16 @@ import Network.Wai.Handler.Warp.Types
2425
2526-- Set-Cookie: contains only one cookie value.
2627-- So, we don't need to split it.
27−hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> H.Status
28− -> H.ResponseHeaders -> IO Builder
29−hpackEncodeHeader ctx ii settings s h = do
28+hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> Response
29+ -> IO Builder
30+hpackEncodeHeader ctx ii settings rsp = do
3031 hdr1 <- addServerAndDate h
3132 let hdr2 = (":status", status) : map (first foldedCase) hdr1
3233 hpackEncodeRawHeaders ctx hdr2
3334 where
34− status = B8.pack $ show $ H.statusCode $ s
35+ s = responseStatus rsp
36+ h = responseHeaders rsp
37+ status = B8.pack $ show $ H.statusCode s
3538 dc = dateCacher ii
3639 rspidxhdr = indexResponseHeader h
3740 defServer = S.settingsServerName settings
@@ -54,20 +57,6 @@ hpackDecodeHeader hdrblk Context{decodeDynamicTable} = do
5457 hdrtbl <- readIORef decodeDynamicTable
5558 (hdrtbl', hdr) <- decodeHeader hdrtbl hdrblk `E.onException` cleanup
5659 writeIORef decodeDynamicTable hdrtbl'
57− return $ concatCookie hdr
60+ return hdr
5861 where
5962 cleanup = E.throwIO $ ConnectionError CompressionError "cannot decompress the header"
60−
61−-- |
62−--
63−-- >>> concatCookie [("foo","bar")]
64−-- [("foo","bar")]
65−-- >>> concatCookie [("cookie","a=b"),("foo","bar"),("cookie","c=d"),("cookie","e=f")]
66−-- [("foo","bar"),("cookie","a=b; c=d; e=f")]
67−concatCookie :: HeaderList -> HeaderList
68−concatCookie = collect []
69− where
70− collect cookies (("cookie",c):rest) = collect (cookies ++ [c]) rest
71− collect cookies (h:rest) = h : collect cookies rest
72− collect [] [] = []
73− collect cookies [] = [("cookie", B.intercalate "; " cookies)]
warp/Network/Wai/Handler/Warp/HTTP2/Manager.hs+31−31
@@ -1,4 +1,4 @@
1−{-# LANGUAGE CPP #-}
1+{-# LANGUAGE CPP, BangPatterns #-}
22
33-- | A thread pool manager.
44-- The manager has responsibility to spawn and kill
@@ -20,42 +20,45 @@ import Control.Concurrent.STM
2020import Control.Monad (void)
2121import Data.Set (Set)
2222import qualified Data.Set as Set
23+import Data.Foldable
2324import Network.Wai.Handler.Warp.IORef
25+import Network.Wai.Handler.Warp.Settings
26+import qualified Network.Wai.Handler.Warp.Timeout as T
2427
2528----------------------------------------------------------------
2629
30+type Action = T.Manager -> IO ()
31+
2732data Command = Stop | Spawn | Replace ThreadId
2833
29−data Manager = Manager (TQueue Command) (IORef (IO ()))
34+data Manager = Manager (TQueue Command) (IORef Action)
3035
3136-- | Starting a thread pool manager.
3237-- Its action is initially set to 'return ()' and should be set
3338-- by 'setAction'. This allows that the action can include
3439-- the manager itself.
35−start :: IO Manager
36−start = do
37− tset <- newThreadSet
40+start :: Settings -> IO Manager
41+start set = do
3842 q <- newTQueueIO
39− ref <- newIORef (return ())
40− void $ forkIO $ go q tset ref
43+ ref <- newIORef (\_ -> return ())
44+ timmgr <- T.initialize $ settingsTimeout set * 1000000
45+ void $ forkIO $ go q Set.empty ref timmgr
4146 return $ Manager q ref
4247 where
43− go q tset ref = do
48+ go q !tset0 ref timmgr = do
4449 x <- atomically $ readTQueue q
4550 case x of
46− Stop -> kill tset
47− Spawn -> next
48− Replace oldtid -> do
49− del tset oldtid
50− next
51+ Stop -> kill tset0 >> T.killManager timmgr
52+ Spawn -> next tset0
53+ Replace oldtid -> next $ del oldtid tset0
5154 where
52− next = do
55+ next tset = do
5356 action <- readIORef ref
54− newtid <- forkIO action
55− add tset newtid
56− go q tset ref
57+ newtid <- forkIO (action timmgr)
58+ let !tset' = add newtid tset
59+ go q tset' ref timmgr
5760
58−setAction :: Manager -> IO () -> IO ()
61+setAction :: Manager -> Action -> IO ()
5962setAction (Manager _ ref) action = writeIORef ref action
6063
6164stop :: Manager -> IO ()
@@ -69,18 +72,15 @@ replaceWithAction (Manager q _) tid = atomically $ writeTQueue q $ Replace tid
6972
7073----------------------------------------------------------------
7174
72−newtype ThreadSet = ThreadSet (IORef (Set ThreadId))
73−
74−newThreadSet :: IO ThreadSet
75−newThreadSet = ThreadSet <$> newIORef Set.empty
76−
77−add :: ThreadSet -> ThreadId -> IO ()
78−add (ThreadSet ref) tid =
79− atomicModifyIORef' ref (\set -> (Set.insert tid set, ()))
75+add :: ThreadId -> Set ThreadId -> Set ThreadId
76+add tid set = set'
77+ where
78+ !set' = Set.insert tid set
8079
81−del :: ThreadSet -> ThreadId -> IO ()
82−del (ThreadSet ref) tid =
83− atomicModifyIORef' ref (\set -> (Set.delete tid set, ()))
80+del :: ThreadId -> Set ThreadId -> Set ThreadId
81+del tid set = set'
82+ where
83+ !set' = Set.delete tid set
8484
85−kill :: ThreadSet -> IO ()
86−kill (ThreadSet ref) = Set.toList <$> readIORef ref >>= mapM_ killThread
85+kill :: Set ThreadId -> IO ()
86+kill set = traverse_ killThread set
warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+15−25
@@ -56,7 +56,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
5656 cont <- processStreamGuardingError $ decodeFrameHeader hd
5757 when cont loop
5858
59− processStreamGuardingError (FrameHeaders, FrameHeader{streamId})
59+ processStreamGuardingError (_, FrameHeader{streamId})
6060 | isResponse streamId = E.throwIO $ ConnectionError ProtocolError "stream id should be odd"
6161 processStreamGuardingError (FrameUnknown _, FrameHeader{payloadLength}) = do
6262 mx <- readIORef continued
@@ -92,7 +92,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
9292 control ftyp header pl ctx
9393 | otherwise = do
9494 checkContinued
95− strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream
95+ !strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream
9696 pl <- recvN payloadLength
9797 state <- readIORef streamState
9898 state' <- stream ftyp header pl ctx state strm
@@ -105,7 +105,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
105105 E.throwIO $ StreamError ProtocolError streamId
106106 writeIORef streamPrecedence $ toPrecedence pri
107107 writeIORef streamState HalfClosed
108− let req = mkreq vh (return "")
108+ let !req = mkreq vh (return "")
109109 atomically $ writeTQueue inputQ $ Input strm req
110110 Nothing -> E.throwIO $ StreamError ProtocolError streamId
111111 Open (HasBody hdr pri) -> do
@@ -118,7 +118,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
118118 writeIORef streamContentLength $ vhCL vh
119119 readQ <- newReadBody q
120120 bodySource <- mkSource readQ
121− let req = mkreq vh (readSource bodySource)
121+ let !req = mkreq vh (readSource bodySource)
122122 atomically $ writeTQueue inputQ $ Input strm req
123123 Nothing -> E.throwIO $ StreamError ProtocolError streamId
124124 s@(Open Continued{}) -> do
@@ -147,29 +147,20 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
147147 when (isHalfClosed st) $ E.throwIO $ ConnectionError StreamClosed "header must not be sent to half closed"
148148 return strm0
149149 Nothing -> do
150+ -- checkme
150151 when (ftyp `notElem` [FrameHeaders,FramePriority]) $
151152 E.throwIO $ ConnectionError ProtocolError "this frame is not allowed in an idel stream"
153+ csid <- readIORef currentStreamId
154+ when (streamId <= csid) $
155+ E.throwIO $ ConnectionError ProtocolError "stream identifier must not decrease"
152156 when (ftyp == FrameHeaders) $ do
153− csid <- readIORef currentStreamId
154− if streamId <= csid then
155− E.throwIO $ ConnectionError ProtocolError "stream identifier must not decrease"
156− else
157− writeIORef currentStreamId streamId
157+ writeIORef currentStreamId streamId
158158 cnt <- readIORef concurrency
159− when (cnt >= recommendedConcurrency) $ do
160− -- Record that the stream is closed, rather than
161− -- idle, so that receiving frames on it is only a
162− -- stream error.
163− consume payloadLength
164− strm <- newStream concurrency streamId 0
165− writeIORef (streamState strm) $ Closed $
166− ResetByMe $ E.toException $
167− StreamError RefusedStream streamId
168− insert streamTable streamId strm
159+ when (cnt >= recommendedConcurrency) $
169160 E.throwIO $ StreamError RefusedStream streamId
170161 ws <- initialWindowSize <$> readIORef http2settings
171− newstrm <- newStream concurrency streamId (fromIntegral ws)
172− when (ftyp == FrameHeaders) $ opened newstrm
162+ newstrm <- newStream streamId (fromIntegral ws)
163+ when (ftyp == FrameHeaders) $ opened ctx newstrm
173164 insert streamTable streamId newstrm
174165 return newstrm
175166
@@ -184,7 +175,7 @@ control FrameSettings header@FrameHeader{flags} bs Context{http2settings, output
184175 Just x -> E.throwIO x
185176 Nothing -> return ()
186177 unless (testAck flags) $ do
187− modifyIORef http2settings $ \old -> updateSettings old alist
178+ modifyIORef' http2settings $ \old -> updateSettings old alist
188179 let frame = settingsFrame setAck []
189180 enqueueControl outputQ 0 $ OSettings frame alist
190181 return True
@@ -311,16 +302,15 @@ stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s Stream{streamWindow
311302 atomically $ writeTVar streamWindow w
312303 return s
313304
314−stream FrameRSTStream header bs _ _ strm = do
305+stream FrameRSTStream header bs ctx _ strm = do
315306 RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs
316307 let cc = Reset e
317− closed strm cc
308+ closed ctx strm cc
318309 return $ Closed cc -- will be written to streamState again
319310
320311stream FramePriority header bs Context{outputQ,priorityTreeSize} s Stream{streamNumber,streamPrecedence} = do
321312 PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs
322313 checkPriority newpri streamNumber
323− -- checkme: this should be tested
324314 oldpre <- readIORef streamPrecedence
325315 let !newpre = toPrecedence newpri
326316 writeIORef streamPrecedence newpre
warp/Network/Wai/Handler/Warp/HTTP2/Request.hs+46−28
@@ -1,4 +1,5 @@
11{-# LANGUAGE OverloadedStrings, CPP #-}
2+{-# LANGUAGE BangPatterns #-}
23
34module Network.Wai.Handler.Warp.HTTP2.Request (
45 mkRequest
@@ -38,12 +39,12 @@ import qualified Network.Wai.Handler.Warp.Settings as S (Settings, settingsNoPar
3839import qualified Network.Wai.Handler.Warp.Timeout as Timeout
3940
4041data ValidHeaders = ValidHeaders {
41− vhMethod :: ByteString
42− , vhPath :: ByteString
43− , vhAuth :: Maybe ByteString
44− , vhCL :: Maybe Int -- ^ Content-Length
45− , vhHeader :: RequestHeaders
46− }
42+ vhMethod :: !ByteString
43+ , vhPath :: !ByteString
44+ , vhAuth :: !(Maybe ByteString)
45+ , vhCL :: !(Maybe Int)
46+ , vhHeader :: !RequestHeaders
47+ } deriving Show
4748
4849type MkReq = ValidHeaders -> IO ByteString -> Request
4950
@@ -51,8 +52,8 @@ mkRequest :: InternalInfo -> S.Settings -> SockAddr -> MkReq
5152mkRequest ii settings addr (ValidHeaders m p ma _ hdr) body = req
5253 where
5354 (unparsedPath,query) = B8.break (=='?') p
54− path = H.extractPath unparsedPath
55− req = Request {
55+ !path = H.extractPath unparsedPath
56+ !req = Request {
5657 requestMethod = m
5758 , httpVersion = http2ver
5859 , rawPathInfo = if S.settingsNoParsePath settings then unparsedPath else path
@@ -70,8 +71,8 @@ mkRequest ii settings addr (ValidHeaders m p ma _ hdr) body = req
7071 , requestHeaderHost = ma
7172 , requestHeaderRange = lookup hRange hdr
7273 }
73− th = threadHandle ii
74− vaultValue = Vault.insert pauseTimeoutKey (Timeout.pause th)
74+ !th = threadHandle ii
75+ !vaultValue = Vault.insert pauseTimeoutKey (Timeout.pause th)
7576 $ Vault.insert getFileInfoKey (fileInfo ii)
7677 Vault.empty
7778
@@ -82,53 +83,70 @@ data Pseudo = Pseudo {
8283 , colonPath :: !(Maybe ByteString)
8384 , colonAuth :: !(Maybe ByteString)
8485 , contentLen :: !(Maybe ByteString)
85− }
86+ } deriving Show
8687
8788emptyPseudo :: Pseudo
8889emptyPseudo = Pseudo Nothing Nothing Nothing Nothing
8990
91+-- |
92+--
93+-- >>> validateHeaders [(":method","GET"),(":path","path")]
94+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Nothing, vhCL = Nothing, vhHeader = []})
95+-- >>> validateHeaders [(":method","GET"),(":path","path"),(":authority","authority"),("accept-language","en")]
96+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Just "authority", vhCL = Nothing, vhHeader = [("accept-language","en")]})
97+-- >>> validateHeaders [(":method","GET"),(":path","path"),("cookie","a=b"),("accept-language","en"),("cookie","c=d"),("cookie","e=f")]
98+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Nothing, vhCL = Nothing, vhHeader = [("accept-language","en"),("cookie","a=b; c=d; e=f")]})
9099validateHeaders :: HeaderList -> Maybe ValidHeaders
91−validateHeaders hs = case pseudo hs (emptyPseudo,id) of
92− Just (Pseudo (Just m) (Just p) ma mcl, h)
93− -> Just $ ValidHeaders m p ma (readInt <$> mcl) h
100+validateHeaders hs = case pseudo hs emptyPseudo of
101+ Just (Pseudo (Just m) (Just p) ma mcl, !h)
102+ -> Just $! ValidHeaders m p ma (readInt <$> mcl) h
94103 _ -> Nothing
95104 where
96− pseudo [] (p,b) = Just (p,b [])
97− pseudo h@((k,v):kvs) (p,b)
105+ pseudo [] !p = Just (p,[])
106+ pseudo h@((k,v):kvs) !p
98107 | k == ":method" = if isJust (colonMethod p) then
99108 Nothing
100109 else
101− pseudo kvs (p { colonMethod = Just v },b)
110+ pseudo kvs (p { colonMethod = Just v })
102111 | k == ":path" = if isJust (colonPath p) then
103112 Nothing
104113 else
105− pseudo kvs (p { colonPath = Just v },b)
114+ pseudo kvs (p { colonPath = Just v })
106115 | k == ":authority" = if isJust (colonAuth p) then
107116 Nothing
108117 else
109− pseudo kvs (p { colonAuth = Just v },b)
110− | k == ":scheme" = pseudo kvs (p,b) -- fixme: how to store :scheme?
118+ pseudo kvs (p { colonAuth = Just v })
119+ | k == ":scheme" = pseudo kvs p -- fixme: how to store :scheme?
111120 | isPseudo k = Nothing
112− | otherwise = normal h (p,b)
121+ | otherwise = normal h (p,id,id)
113122
114− normal [] (p,b) = Just (p,b [])
115− normal ((k,v):kvs) (p,b)
123+ normal [] (!p,b,c) = Just (p, mkH b c)
124+ normal ((k,v):kvs) (!p,b,c)
116125 | isPseudo k = Nothing
117126 | k == "connection" = Nothing
118127 | k == "te" = if v == "trailers" then
119− normal kvs (p, b . ((mk k,v) :))
128+ normal kvs (p, b . ((mk k,v) :), c)
120129 else
121130 Nothing
122131 | k == "content-length"
123− = normal kvs (p { contentLen = Just v }, b . ((mk k,v) :))
132+ = normal kvs (p { contentLen = Just v }, b . ((mk k,v) :), c)
124133 | k == "host" = if isJust (colonAuth p) then
125− normal kvs (p,b)
134+ normal kvs (p, b, c)
126135 else
127− normal kvs (p { colonAuth = Just v },b)
136+ normal kvs (p { colonAuth = Just v }, b, c)
137+ | k == "cookie" = normal kvs (p, b, c . (v:))
128138 | otherwise = case BS.find isUpper k of
129− Nothing -> normal kvs (p, b . ((mk k,v) :))
139+ Nothing -> normal kvs (p, b . ((mk k,v) :), c)
130140 Just _ -> Nothing
131141
142+ mkH b c = h
143+ where
144+ !h = b anchor
145+ !cookieList = c []
146+ !anchor
147+ | null cookieList = []
148+ | otherwise = let !v = BS.intercalate "; " cookieList
149+ in [("cookie",v)]
132150 isPseudo "" = False
133151 isPseudo k = BS.head k == _colon
134152
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+238−228
@@ -7,71 +7,55 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
77#if __GLASGOW_HASKELL__ < 709
88import Control.Applicative
99#endif
10−import Control.Concurrent.MVar (putMVar)
1110import Control.Concurrent.STM
1211import qualified Control.Exception as E
13−import Control.Monad (void, when)
12+import Control.Monad (unless, void, when)
1413import qualified Data.ByteString as BS
15−import qualified Data.ByteString.Builder as B (int32BE)
1614import qualified Data.ByteString.Builder.Extra as B
17−import Data.Monoid ((<>))
1815import Foreign.Ptr
19−import qualified Network.HTTP.Types as H
2016import Network.HPACK (setLimitForEncoding)
2117import Network.HTTP2
2218import Network.HTTP2.Priority
23−import Network.Wai (FilePart(..))
24−import Network.Wai.HTTP2 (Trailers, promiseHeaders)
19+import Network.Wai
2520import Network.Wai.Handler.Warp.Buffer
2621import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
2722import Network.Wai.Handler.Warp.HTTP2.HPACK
2823import Network.Wai.Handler.Warp.HTTP2.Types
2924import Network.Wai.Handler.Warp.IORef
3025import qualified Network.Wai.Handler.Warp.Settings as S
26+import qualified Network.Wai.Handler.Warp.Timeout as T
3127import Network.Wai.Handler.Warp.Types
28+import Network.Wai.Internal (Response(..))
3229
3330#ifdef WINDOWS
3431import qualified System.IO as IO
3532#else
36−import Network.Wai.Handler.Warp.FdCache (getFd)
33+import Network.Wai.Handler.Warp.FdCache
3734import Network.Wai.Handler.Warp.SendFile (positionRead)
38−import qualified Network.Wai.Handler.Warp.Timeout as T
3935import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)
40−import System.Posix.Types (Fd)
36+import System.Posix.Types
4137#endif
4238
4339----------------------------------------------------------------
4440
45−-- | The platform-specific type of an open file to stream from. On Windows we
46−-- don't have pread, so this is just a Handle; on Unix platforms with pread,
47−-- this is a file descriptor supplied by the fd cache.
48−#ifdef WINDOWS
49−type OpenFile = IO.Handle
50−#else
51−type OpenFile = Fd
52−#endif
53−
5441data Leftover = LZero
5542 | LOne B.BufferWriter
5643 | LTwo BS.ByteString B.BufferWriter
57− | LFile OpenFile Integer Integer (IO ())
5844
5945----------------------------------------------------------------
6046
61−-- | Run the given action if the stream is not closed; handle any exceptions by
62−-- resetting the stream.
63−unlessClosed :: Connection -> Stream -> IO () -> IO Bool
64−unlessClosed Connection{connSendAll}
47+unlessClosed :: Context -> Connection -> Stream -> IO () -> IO ()
48+unlessClosed ctx
49+ Connection{connSendAll}
6550 strm@Stream{streamState,streamNumber}
6651 body = E.handle resetStream $ do
6752 state <- readIORef streamState
68− if (isClosed state) then return False else body >> return True
53+ unless (isClosed state) body
6954 where
7055 resetStream e = do
71− closed strm (ResetByMe e)
56+ closed ctx strm (ResetByMe e)
7257 let rst = resetFrame InternalError streamNumber
7358 connSendAll rst
74− return False
7559
7660getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
7761getWindowSize connWindow strmWindow = do
@@ -106,6 +90,9 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
10690
10791 switch OFinish _ = return ()
10892 switch (OGoaway frame) _ = connSendAll frame
93+ switch (OFrame frame) _ = do
94+ connSendAll frame
95+ loop
10996 switch (OSettings frame alist) _ = do
11097 connSendAll frame
11198 case lookup SettingsHeaderTableSize alist of
@@ -114,104 +101,66 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
114101 dyntbl <- readIORef encodeDynamicTable
115102 setLimitForEncoding siz dyntbl
116103 loop
117− switch (OFrame frame) _ = do
118− connSendAll frame
119− loop
120− switch (OResponse strm s h aux) pre = do
121− writeIORef (streamPrecedence strm) pre -- fixme
122− _ <- unlessClosed conn strm $
123− getWindowSize connectionWindow (streamWindow strm) >>=
124− sendResponse strm s h aux
104+ switch (OResponse strm rsp aux) pre = do
105+ writeIORef (streamPrecedence strm) pre
106+ unlessClosed ctx conn strm $ do
107+ lim <- getWindowSize connectionWindow (streamWindow strm)
108+ -- Header frame and Continuation frame
109+ let sid = streamNumber strm
110+ endOfStream = case aux of
111+ Persist{} -> False
112+ Oneshot hb -> not hb
113+ len <- headerContinue sid rsp endOfStream
114+ let total = len + frameHeaderLength
115+ case aux of
116+ Oneshot True -> do -- hasBody
117+ -- Data frame payload
118+ (off, _) <- sendHeadersIfNecessary total
119+ let payloadOff = off + frameHeaderLength
120+ Next datPayloadLen mnext <-
121+ fillResponseBodyGetNext conn ii payloadOff lim rsp
122+ fillDataHeaderSend strm total datPayloadLen mnext
123+ maybeEnqueueNext strm mnext
124+ Oneshot False -> do
125+ -- "closed" must be before "connSendAll". If not,
126+ -- the context would be switched to the receiver,
127+ -- resulting the inconsistency of concurrency.
128+ closed ctx strm Finished
129+ flushN total
130+ Persist sq tvar -> do
131+ (off, needSend) <- sendHeadersIfNecessary total
132+ let payloadOff = off + frameHeaderLength
133+ Next datPayloadLen mnext <-
134+ fillStreamBodyGetNext conn payloadOff lim sq tvar strm
135+ -- If no data was immediately available, avoid sending an
136+ -- empty data frame.
137+ if datPayloadLen > 0 then
138+ fillDataHeaderSend strm total datPayloadLen mnext
139+ else
140+ when needSend $ flushN off
141+ maybeEnqueueNext strm mnext
125142 loop
126143 switch (ONext strm curr) pre = do
127144 writeIORef (streamPrecedence strm) pre
128− _ <- unlessClosed conn strm $ do
145+ unlessClosed ctx conn strm $ do
129146 lim <- getWindowSize connectionWindow (streamWindow strm)
130147 -- Data frame payload
131148 Next datPayloadLen mnext <- curr lim
132− fillDataHeaderSend strm 0 datPayloadLen
133− dispatchNext strm mnext
134− loop
135− switch (OPush oldStrm push mvar strm s h aux) pre = do
136− writeIORef (streamPrecedence strm) pre -- fixme
137− pushed <- unlessClosed conn oldStrm $ do
138− lim <- getWindowSize connectionWindow (streamWindow strm)
139− -- Write and send the promise.
140− builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push
141− off <- pushContinue (streamNumber oldStrm) (streamNumber strm) builder
142− flushN $ off + frameHeaderLength
143− -- TODO(awpr): refactor sendResponse to be able to handle non-zero
144− -- initial offsets and use that to potentially avoid the extra syscall.
145− sendResponse strm s h aux lim
146− putMVar mvar pushed
149+ fillDataHeaderSend strm 0 datPayloadLen mnext
150+ maybeEnqueueNext strm mnext
147151 loop
148152
149− -- Send the response headers and as much of the response as is immediately
150− -- available; shared by normal responses and pushed streams.
151− sendResponse :: Stream -> H.Status -> H.ResponseHeaders -> Aux -> WindowSize -> IO ()
152− sendResponse strm s h (Persist sq tvar) lim = do
153− -- Header frame and Continuation frame
154− let sid = streamNumber strm
155− builder <- hpackEncodeHeader ctx ii settings s h
156− len <- headerContinue sid builder False
157− let total = len + frameHeaderLength
158− (off, needSend) <- sendHeadersIfNecessary total
159− let payloadOff = off + frameHeaderLength
160− Next datPayloadLen mnext <-
161− fillStreamBodyGetNext ii conn payloadOff lim sq tvar strm
162− -- If no data was immediately available, avoid sending an
163− -- empty data frame.
164− if datPayloadLen > 0 then
165− fillDataHeaderSend strm total datPayloadLen
166− else
167− when needSend $ flushN off
168− dispatchNext strm mnext
169−
170− -- Send the stream's trailers and close the stream.
171− sendTrailers :: Stream -> Trailers -> IO ()
172− sendTrailers strm trailers = do
173− -- Trailers always indicate the end of a stream; send them in
174− -- consecutive header+continuation frames and end the stream. Some
175− -- clients dislike empty headers frames, so end the stream with an
176− -- empty data frame instead, as recommended by the spec.
177− toFlush <- case trailers of
178− [] -> frameHeaderLength <$ fillFrameHeader FrameData 0
179− (streamNumber strm)
180− (setEndStream defaultFlags)
181− connWriteBuffer
182− _ -> do
183− builder <- hpackEncodeCIHeaders ctx trailers
184− off <- headerContinue (streamNumber strm) builder True
185− return (off + frameHeaderLength)
186− -- 'closed' must be before 'flushN'. If not, the context would be
187− -- switched to the receiver, resulting in the inconsistency of
188− -- concurrency.
189− closed strm Finished
190− flushN toFlush
191−
192153 -- Flush the connection buffer to the socket, where the first 'n' bytes of
193154 -- the buffer are filled.
194155 flushN :: Int -> IO ()
195156 flushN n = bufferIO connWriteBuffer n connSendAll
196157
197− -- A flags value with the end-header flag set iff the argument is B.Done.
198− maybeEndHeaders B.Done = setEndHeader defaultFlags
199− maybeEndHeaders _ = defaultFlags
200−
201− -- Write PUSH_PROMISE and possibly CONTINUATION frames into the connection
202− -- buffer, using the given builder as their contents; flush them to the
203− -- socket as necessary.
204− pushContinue sid newSid builder = do
205− let builder' = B.int32BE (fromIntegral newSid) <> builder
206− (len, signal) <- B.runBuilder builder' bufHeaderPayload headerPayloadLim
207− let flag = maybeEndHeaders signal
208− fillFrameHeader FramePushPromise len sid flag connWriteBuffer
209− continue sid len signal
210−
211− -- Write HEADER and possibly CONTINUATION frames.
212− headerContinue sid builder endOfStream = do
158+ headerContinue sid rsp endOfStream = do
159+ builder <- hpackEncodeHeader ctx ii settings rsp
213160 (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim
214− let flag0 = maybeEndHeaders signal
161+ let flag0 = case signal of
162+ B.Done -> setEndHeader defaultFlags
163+ _ -> defaultFlags
215164 flag = if endOfStream then setEndStream flag0 else flag0
216165 fillFrameHeader FrameHeaders len sid flag connWriteBuffer
217166 continue sid len signal
@@ -220,7 +169,9 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
220169 continue sid len (B.More _ writer) = do
221170 flushN $ len + frameHeaderLength
222171 (len', signal') <- writer bufHeaderPayload headerPayloadLim
223− let flag = maybeEndHeaders signal'
172+ let flag = case signal' of
173+ B.Done -> setEndHeader defaultFlags
174+ _ -> defaultFlags
224175 fillFrameHeader FrameContinuation len' sid flag connWriteBuffer
225176 continue sid len' signal'
226177 continue sid len (B.Chunk bs writer) = do
@@ -237,22 +188,16 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
237188 -- True if the connection buffer has room for a 1-byte data frame.
238189 canFitDataFrame total = total + frameHeaderLength < connBufferSize
239190
240− -- Take the appropriate action based on the given 'Control':
241− -- - If more output is immediately available, re-enqueue the stream in the
242− -- output queue.
243− -- - If the output is over and trailers are available, send them now and
244− -- end the stream.
245− -- - If we've drained the queue and handed the stream back to its waiter,
246− -- do nothing.
247− --
248− -- This is done after sending any part of the stream body, so it's shared
249− -- by 'sendResponse' and @switch (ONext ...)@.
250− dispatchNext :: Stream -> Control DynaNext -> IO ()
251− dispatchNext _ CNone = return ()
252− dispatchNext strm (CFinish trailers) = sendTrailers strm trailers
253− dispatchNext strm (CNext next) = do
191+ -- Re-enqueue the stream in the output queue if more output is immediately
192+ -- available; do nothing otherwise.
193+ maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
194+ maybeEnqueueNext strm (CNext next) = do
254195 let out = ONext strm next
255196 enqueueOrSpawnTemporaryWaiter strm outputQ out
197+ -- If the streaming is not finished, it must already have been
198+ -- written to the 'TVar' owned by 'waiter', which will
199+ -- put it back into the queue when more output becomes available.
200+ maybeEnqueueNext _ _ = return ()
256201
257202
258203 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -264,12 +209,21 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
264209 flushN total
265210 return (0, False)
266211
267− fillDataHeaderSend strm otherLen datPayloadLen = do
212+ fillDataHeaderSend strm otherLen datPayloadLen mnext = do
268213 -- Data frame header
269214 let sid = streamNumber strm
270215 buf = connWriteBuffer `plusPtr` otherLen
271216 total = otherLen + frameHeaderLength + datPayloadLen
272− fillFrameHeader FrameData datPayloadLen sid defaultFlags buf
217+ flag = case mnext of
218+ CFinish -> setEndStream defaultFlags
219+ _ -> defaultFlags
220+ fillFrameHeader FrameData datPayloadLen sid flag buf
221+ -- "closed" must be before "connSendAll". If not,
222+ -- the context would be switched to the receiver,
223+ -- resulting the inconsistency of concurrency.
224+ case mnext of
225+ CFinish -> closed ctx strm Finished
226+ _ -> return ()
273227 flushN total
274228 atomically $ do
275229 modifyTVar' connectionWindow (subtract datPayloadLen)
@@ -281,108 +235,128 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
281235
282236----------------------------------------------------------------
283237
284−fillStreamBodyGetNext :: InternalInfo -> Connection -> Int -> WindowSize
285− -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
286−fillStreamBodyGetNext ii Connection{connWriteBuffer,connBufferSize}
287− off lim sq tvar strm = do
238+{-
239+ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)
240+ResponseBuilder Status ResponseHeaders Builder
241+ResponseStream Status ResponseHeaders StreamingBody
242+ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response
243+-}
244+
245+fillResponseBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Response -> IO Next
246+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
247+ _ off lim (ResponseBuilder _ _ bb) = do
288248 let datBuf = connWriteBuffer `plusPtr` off
289249 room = min (connBufferSize - off) lim
290− (leftover, cont, len) <- runStreamBuilder ii datBuf room sq
291− nextForStream ii connWriteBuffer connBufferSize sq tvar strm leftover cont len
292−
293−----------------------------------------------------------------
294−
295−runStreamBuilder :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence
296− -> IO (Leftover, Maybe Trailers, BytesFilled)
297−runStreamBuilder ii buf0 room0 sq = loop buf0 room0 0
298− where
299− loop !buf !room !total = do
300− mbuilder <- atomically $ tryReadTBQueue sq
301− case mbuilder of
302− Nothing -> return (LZero, Nothing, total)
303− Just (SBuilder builder) -> do
304− (len, signal) <- B.runBuilder builder buf room
305− let !total' = total + len
306− case signal of
307− B.Done -> loop (buf `plusPtr` len) (room - len) total'
308− B.More _ writer -> return (LOne writer, Nothing, total')
309− B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')
310− Just (SFile path part) -> do
311− (leftover, len) <- runStreamFile ii buf room path part
312− let !total' = total + len
313− case leftover of
314− LZero -> loop (buf `plusPtr` len) (room - len) total'
315− _ -> return (leftover, Nothing, total')
316−
317− Just SFlush -> return (LZero, Nothing, total)
318− Just (SFinish trailers) -> return (LZero, Just trailers, total)
319−
320−-- | Open the file and start reading into the send buffer.
321−runStreamFile :: InternalInfo -> Buffer -> BufSize -> FilePath -> FilePart
322− -> IO (Leftover, BytesFilled)
323−
324−-- | Read the given (OS-specific) file representation into the buffer. On
325−-- non-Windows systems this uses pread; on Windows this ignores the position
326−-- because we use the Handle's internal read position instead (because it's not
327−-- potentially shared with other readers).
328−readOpenFile :: OpenFile -> Buffer -> BufSize -> Integer -> IO Int
250+ (len, signal) <- B.runBuilder bb datBuf room
251+ return $ nextForBuilder connWriteBuffer connBufferSize len signal
329252
330253#ifdef WINDOWS
331−runStreamFile _ buf room path part = do
332− let start = filePartOffset part
333− bytes = filePartByteCount part
254+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
255+ _ off lim (ResponseFile _ _ path mpart) = do
256+ let datBuf = connWriteBuffer `plusPtr` off
257+ room = min (connBufferSize - off) lim
258+ (start, bytes) <- fileStartEnd path mpart
334259 -- fixme: how to close Handle? GC does it at this moment.
335260 h <- IO.openBinaryFile path IO.ReadMode
336261 IO.hSeek h IO.AbsoluteSeek start
337− fillBufFile buf room h start bytes (return ())
338−
339−readOpenFile h buf room _ = IO.hGetBufSome h buf room
262+ len <- IO.hGetBufSome h datBuf (mini room bytes)
263+ let bytes' = bytes - fromIntegral len
264+ return $ nextForFile len connWriteBuffer connBufferSize h bytes' (return ())
340265#else
341−runStreamFile ii buf room path part = do
342− let start = filePartOffset part
343− bytes = filePartByteCount part
266+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
267+ ii off lim (ResponseFile _ _ path mpart) = do
344268 (fd, refresh) <- case fdCacher ii of
345269 Just fdcache -> getFd fdcache path
346270 Nothing -> do
347271 fd' <- openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True}
348272 th <- T.register (timeoutManager ii) (closeFd fd')
349273 return (fd', T.tickle th)
350− fillBufFile buf room fd start bytes refresh
351−
352−readOpenFile = positionRead
353−#endif
354−
355−-- | Read as much of the file as is currently available into the buffer, then
356−-- return a 'Leftover' to indicate whether this file chunk has more data to
357−-- send. If this read hit the end of the file range, return 'LZero'; otherwise
358−-- return 'LFile' so this stream will continue reading from the file the next
359−-- time it's pulled from the queue.
360−fillBufFile :: Buffer -> BufSize -> OpenFile -> Integer -> Integer -> (IO ())
361− -> IO (Leftover, BytesFilled)
362−fillBufFile buf room f start bytes refresh = do
363− len <- readOpenFile f buf (mini room bytes) start
274+ let datBuf = connWriteBuffer `plusPtr` off
275+ room = min (connBufferSize - off) lim
276+ (start, bytes) <- fileStartEnd path mpart
277+ len <- positionRead fd datBuf (mini room bytes) start
364278 refresh
365279 let len' = fromIntegral len
366− leftover = if bytes > len' then
367− LFile f (start + len') (bytes - len') refresh
368− else
369− LZero
370− return (leftover, len)
280+ return $ nextForFile len connWriteBuffer connBufferSize fd (start + len') (bytes - len') refresh
281+#endif
371282
372−mini :: Int -> Integer -> Int
373−mini i n
374− | fromIntegral i < n = i
375− | otherwise = fromIntegral n
283+fillResponseBodyGetNext _ _ _ _ _ = error "fillResponseBodyGetNext"
284+
285+fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)
286+fileStartEnd _ (Just part) =
287+ return (filePartOffset part, filePartByteCount part)
288+fileStartEnd _ _ = error "fileStartEnd"
289+
290+----------------------------------------------------------------
291+
292+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
293+fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
294+ off lim sq tvar strm = do
295+ let datBuf = connWriteBuffer `plusPtr` off
296+ room = min (connBufferSize - off) lim
297+ (leftover, cont, len) <- runStreamBuilder datBuf room sq
298+ nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
299+
300+----------------------------------------------------------------
301+
302+fillBufBuilder :: Buffer -> BufSize -> Leftover -> DynaNext
303+fillBufBuilder buf0 siz0 leftover lim = do
304+ let payloadBuf = buf0 `plusPtr` frameHeaderLength
305+ room = min (siz0 - frameHeaderLength) lim
306+ case leftover of
307+ LZero -> error "fillBufBuilder: LZero"
308+ LOne writer -> do
309+ (len, signal) <- writer payloadBuf room
310+ getNext len signal
311+ LTwo bs writer
312+ | BS.length bs <= room -> do
313+ buf1 <- copy payloadBuf bs
314+ let len1 = BS.length bs
315+ (len2, signal) <- writer buf1 (room - len1)
316+ getNext (len1 + len2) signal
317+ | otherwise -> do
318+ let (bs1,bs2) = BS.splitAt room bs
319+ void $ copy payloadBuf bs1
320+ getNext room (B.Chunk bs2 writer)
321+ where
322+ getNext l s = return $ nextForBuilder buf0 siz0 l s
323+
324+nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next
325+nextForBuilder _ _ len B.Done
326+ = Next len CFinish
327+nextForBuilder buf siz len (B.More _ writer)
328+ = Next len (CNext (fillBufBuilder buf siz (LOne writer)))
329+nextForBuilder buf siz len (B.Chunk bs writer)
330+ = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer)))
331+
332+----------------------------------------------------------------
333+
334+runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence
335+ -> IO (Leftover, Bool, BytesFilled)
336+runStreamBuilder buf0 room0 sq = loop buf0 room0 0
337+ where
338+ loop !buf !room !total = do
339+ mbuilder <- atomically $ tryReadTBQueue sq
340+ case mbuilder of
341+ Nothing -> return (LZero, True, total)
342+ Just (SBuilder builder) -> do
343+ (len, signal) <- B.runBuilder builder buf room
344+ let !total' = total + len
345+ case signal of
346+ B.Done -> loop (buf `plusPtr` len) (room - len) total'
347+ B.More _ writer -> return (LOne writer, True, total')
348+ B.Chunk bs writer -> return (LTwo bs writer, True, total')
349+ Just SFlush -> return (LZero, True, total)
350+ Just SFinish -> return (LZero, False, total)
376351
377−fillBufStream :: InternalInfo -> Buffer -> BufSize -> Leftover
378− -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
379−fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do
352+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
353+fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
380354 let payloadBuf = buf0 `plusPtr` frameHeaderLength
381355 room0 = min (siz0 - frameHeaderLength) lim0
382356 case leftover0 of
383357 LZero -> do
384− (leftover, end, len) <- runStreamBuilder ii payloadBuf room0 sq
385− getNext leftover end len
358+ (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq
359+ getNext leftover cont len
386360 LOne writer -> write writer payloadBuf room0 0
387361 LTwo bs writer
388362 | BS.length bs <= room0 -> do
@@ -392,35 +366,71 @@ fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do
392366 | otherwise -> do
393367 let (bs1,bs2) = BS.splitAt room0 bs
394368 void $ copy payloadBuf bs1
395− getNext (LTwo bs2 writer) Nothing room0
396− LFile fd start bytes refresh -> do
397− (leftover, len) <- fillBufFile payloadBuf room0 fd start bytes refresh
398− getNext leftover Nothing len
369+ getNext (LTwo bs2 writer) True room0
399370 where
400− getNext = nextForStream ii buf0 siz0 sq tvar strm
371+ getNext = nextForStream buf0 siz0 sq tvar strm
401372 write writer1 buf room sofar = do
402373 (len, signal) <- writer1 buf room
403374 case signal of
404375 B.Done -> do
405− (leftover, end, extra) <- runStreamBuilder ii (buf `plusPtr` len) (room - len) sq
376+ (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq
406377 let !total = sofar + len + extra
407− getNext leftover end total
378+ getNext leftover cont total
408379 B.More _ writer -> do
409380 let !total = sofar + len
410− getNext (LOne writer) Nothing total
381+ getNext (LOne writer) True total
411382 B.Chunk bs writer -> do
412383 let !total = sofar + len
413− getNext (LTwo bs writer) Nothing total
414−
415−nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence
416− -> TVar Sync -> Stream -> Leftover -> Maybe Trailers
417− -> BytesFilled -> IO Next
418−nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do
419− atomically $ writeTVar tvar $ SyncFinish
420− return $ Next len $ CFinish trailers
421−nextForStream ii buf siz sq tvar strm LZero Nothing len = do
422− let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm)
384+ getNext (LTwo bs writer) True total
385+
386+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
387+ -> Leftover -> Bool -> BytesFilled
388+ -> IO Next
389+nextForStream _ _ _ tvar _ _ False len = do
390+ atomically $ writeTVar tvar SyncFinish
391+ return $ Next len CFinish
392+nextForStream buf siz sq tvar strm LZero True len = do
393+ let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
423394 atomically $ writeTVar tvar $ SyncNext out
424395 return $ Next len CNone
425−nextForStream ii buf siz sq tvar strm leftover Nothing len =
426− return $ Next len (CNext (fillBufStream ii buf siz leftover sq tvar strm))
396+nextForStream buf siz sq tvar strm leftover True len =
397+ return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
398+
399+----------------------------------------------------------------
400+
401+#ifdef WINDOWS
402+fillBufFile :: Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> DynaNext
403+fillBufFile buf siz h bytes refresh lim = do
404+ let payloadBuf = buf `plusPtr` frameHeaderLength
405+ room = min (siz - frameHeaderLength) lim
406+ len <- IO.hGetBufSome h payloadBuf room
407+ refresh
408+ let bytes' = bytes - fromIntegral len
409+ return $ nextForFile len buf siz h bytes' refresh
410+
411+nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next
412+nextForFile 0 _ _ _ _ _ = Next 0 CFinish
413+nextForFile len _ _ _ 0 _ = Next len CFinish
414+nextForFile len buf siz h bytes refresh =
415+ Next len (CNext (fillBufFile buf siz h bytes refresh))
416+#else
417+fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext
418+fillBufFile buf siz fd start bytes refresh lim = do
419+ let payloadBuf = buf `plusPtr` frameHeaderLength
420+ room = min (siz - frameHeaderLength) lim
421+ len <- positionRead fd payloadBuf (mini room bytes) start
422+ let len' = fromIntegral len
423+ refresh
424+ return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh
425+
426+nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next
427+nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish
428+nextForFile len _ _ _ _ 0 _ = Next len CFinish
429+nextForFile len buf siz fd start bytes refresh =
430+ Next len (CNext (fillBufFile buf siz fd start bytes refresh))
431+#endif
432+
433+mini :: Int -> Integer -> Int
434+mini i n
435+ | fromIntegral i < n = i
436+ | otherwise = fromIntegral n
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+75−141
@@ -1,25 +1,23 @@
11{-# LANGUAGE OverloadedStrings, CPP #-}
22{-# LANGUAGE NamedFieldPuns, RecordWildCards #-}
3+{-# LANGUAGE BangPatterns #-}
34
45module Network.Wai.Handler.Warp.HTTP2.Types where
56
67import Data.ByteString.Builder (Builder)
78#if __GLASGOW_HASKELL__ < 709
8−import Control.Applicative ((<$>), (<*>), pure)
9+import Control.Applicative ((<$>),(<*>))
910#endif
1011import Control.Concurrent (forkIO)
11−import Control.Concurrent.MVar (MVar)
1212import Control.Concurrent.STM
1313import Control.Exception (SomeException)
1414import Control.Monad (void)
15−import Control.Reaper
1615import Data.ByteString (ByteString)
1716import qualified Data.ByteString as BS
1817import Data.IntMap.Strict (IntMap, IntMap)
1918import qualified Data.IntMap.Strict as M
2019import qualified Network.HTTP.Types as H
21−import Network.Wai (Request, FilePart)
22−import Network.Wai.HTTP2 (PushPromise, Trailers)
20+import Network.Wai (Request, Response)
2321import Network.Wai.Handler.Warp.IORef
2422import Network.Wai.Handler.Warp.Types
2523
@@ -38,7 +36,7 @@ isHTTP2 tls = useHTTP2
3836 where
3937 useHTTP2 = case tlsNegotiatedProtocol tls of
4038 Nothing -> False
41− Just proto -> "h2-" `BS.isPrefixOf` proto || proto == "h2"
39+ Just proto -> "h2-" `BS.isPrefixOf` proto
4240
4341----------------------------------------------------------------
4442
@@ -46,118 +44,76 @@ data Input = Input Stream Request
4644
4745----------------------------------------------------------------
4846
49−-- | The result of writing data from a stream's queue into the buffer.
50−data Control a = CFinish Trailers
51− -- ^ The stream has ended, and the trailers should be sent.
47+data Control a = CFinish
5248 | CNext a
53− -- ^ The stream has more data immediately available, and we
54− -- should re-enqueue it when the stream window becomes open.
5549 | CNone
56− -- ^ The stream queue has been drained and we've handed it off
57− -- to its dedicated waiter thread, which will re-enqueue it when
58− -- more data is available.
5950
6051instance Show (Control a) where
61− show (CFinish _) = "CFinish"
62− show (CNext _) = "CNext"
63− show CNone = "CNone"
52+ show CFinish = "CFinish"
53+ show (CNext _) = "CNext"
54+ show CNone = "CNone"
6455
6556type DynaNext = WindowSize -> IO Next
6657
6758type BytesFilled = Int
6859
69−data Next = Next BytesFilled (Control DynaNext)
60+data Next = Next !BytesFilled (Control DynaNext)
7061
7162data Output = OFinish
72− -- ^ Terminate the connection.
73− | OGoaway ByteString
74− -- ^ Send a goaway frame and terminate the connection.
75− | OSettings ByteString SettingsList
76− -- ^ Update settings and send an ack settings frame.
77− | OFrame ByteString
78− -- ^ Send an entire pre-encoded frame.
79− | OResponse Stream H.Status H.ResponseHeaders Aux
80− -- ^ Send the headers and as much of the response as is immediately
81− -- available.
82− | OPush Stream PushPromise (MVar Bool) Stream H.Status H.ResponseHeaders Aux
83− -- ^ Send a PUSH_PROMISE frame, then act like OResponse; signal the
84− -- MVar whether the promise has been sent.
85− | ONext Stream DynaNext
86− -- ^ Send a chunk of the response.
63+ | OGoaway !ByteString
64+ | OFrame !ByteString
65+ | OSettings !ByteString !SettingsList
66+ | OResponse !Stream !Response !Aux
67+ | ONext !Stream !DynaNext
8768
8869outputStream :: Output -> Stream
89−outputStream (OResponse strm _ _ _) = strm
90−outputStream (ONext strm _) = strm
91−outputStream (OPush strm _ _ _ _ _ _) = strm
92−outputStream _ = error "outputStream"
70+outputStream (OResponse strm _ _) = strm
71+outputStream (ONext strm _) = strm
72+outputStream _ = error "outputStream"
9373
9474----------------------------------------------------------------
9575
96−-- | An element on the queue between a running stream and the sender; the order
97−-- should consist of any number of 'SFile', 'SBuilder', and 'SFlush', followed
98−-- by a single 'SFinish'.
99−data Sequence = SFinish Trailers
100− -- ^ The stream is over; its trailers are provided.
76+data Sequence = SFinish
10177 | SFlush
102− -- ^ Any buffered data should be sent immediately.
10378 | SBuilder Builder
104− -- ^ Append a chunk of data to the stream.
105− | SFile FilePath FilePart
106− -- ^ Append a chunk of a file's contents to the stream.
10779
108−-- | A message from the sender to a stream's dedicated waiter thread.
10980data Sync = SyncNone
110− -- ^ Nothing interesting has happened. Go back to sleep.
11181 | SyncFinish
112− -- ^ The stream has ended.
11382 | SyncNext Output
114− -- ^ The stream's queue has been drained; wait for more to be
115− -- available and re-enqueue the given 'Output'.
11683
117−-- | Auxiliary information needed to communicate with a running stream: a queue
118−-- of stream elements ('Sequence') and a 'TVar' connected to its waiter thread.
119−data Aux = Persist (TBQueue Sequence) (TVar Sync)
84+data Aux = Oneshot Bool
85+ | Persist (TBQueue Sequence) (TVar Sync)
12086
12187----------------------------------------------------------------
12288
12389-- | The context for HTTP/2 connection.
12490data Context = Context {
125− http2settings :: IORef Settings
126− , streamTable :: StreamTable
127− -- | Number of active streams initiated by the client; for enforcing our own
128− -- max concurrency setting.
129− , concurrency :: IORef Int
130− -- | Number of active streams initiated by the server; for respecting the
131− -- client's max concurrency setting.
132− , pushConcurrency :: IORef Int
133− , priorityTreeSize :: IORef Int
91+ http2settings :: !(IORef Settings)
92+ , streamTable :: !StreamTable
93+ , concurrency :: !(IORef Int)
94+ , priorityTreeSize :: !(IORef Int)
13495 -- | RFC 7540 says "Other frames (from any stream) MUST NOT
13596 -- occur between the HEADERS frame and any CONTINUATION
13697 -- frames that might follow". This field is used to implement
13798 -- this requirement.
138− , continued :: IORef (Maybe StreamId)
139− , currentStreamId :: IORef StreamId
140− -- ^ Last client-initiated stream ID we've handled.
141− , nextPushStreamId :: IORef StreamId
142− -- ^ Next available server-initiated stream ID.
143− , inputQ :: TQueue Input
144− , outputQ :: PriorityTree Output
145− , encodeDynamicTable :: IORef DynamicTable
146− , decodeDynamicTable :: IORef DynamicTable
147− , connectionWindow :: TVar WindowSize
99+ , continued :: !(IORef (Maybe StreamId))
100+ , currentStreamId :: !(IORef StreamId)
101+ , inputQ :: !(TQueue Input)
102+ , outputQ :: !(PriorityTree Output)
103+ , encodeDynamicTable :: !(IORef DynamicTable)
104+ , decodeDynamicTable :: !(IORef DynamicTable)
105+ , connectionWindow :: !(TVar WindowSize)
148106 }
149107
150108----------------------------------------------------------------
151109
152110newContext :: IO Context
153111newContext = Context <$> newIORef defaultSettings
154− <*> initialize 10 -- fixme: hard coding: 10
155− <*> newIORef 0
112+ <*> newStreamTable
156113 <*> newIORef 0
157114 <*> newIORef 0
158115 <*> newIORef Nothing
159116 <*> newIORef 0
160− <*> newIORef 2 -- first server push stream; 0 is reserved
161117 <*> newTQueueIO
162118 <*> newPriorityTree
163119 <*> (newDynamicTableForEncoding defaultDynamicTableSize >>= newIORef)
@@ -165,32 +121,32 @@ newContext = Context <$> newIORef defaultSettings
165121 <*> newTVarIO defaultInitialWindowSize
166122
167123clearContext :: Context -> IO ()
168−clearContext ctx = void $ reaperStop $ streamTable ctx
124+clearContext _ctx = return ()
169125
170126----------------------------------------------------------------
171127
172128data OpenState =
173129 JustOpened
174130 | Continued [HeaderBlockFragment]
175− Int -- Total size
176− Int -- The number of continuation frames
177− Bool -- End of stream
178− Priority
179− | NoBody HeaderList Priority
180− | HasBody HeaderList Priority
181− | Body (TQueue ByteString)
131+ !Int -- Total size
132+ !Int -- The number of continuation frames
133+ !Bool -- End of stream
134+ !Priority
135+ | NoBody HeaderList !Priority
136+ | HasBody HeaderList !Priority
137+ | Body !(TQueue ByteString)
182138
183139data ClosedCode = Finished
184140 | Killed
185− | Reset ErrorCodeId
141+ | Reset !ErrorCodeId
186142 | ResetByMe SomeException
187143 deriving Show
188144
189145data StreamState =
190146 Idle
191− | Open OpenState
147+ | Open !OpenState
192148 | HalfClosed
193− | Closed ClosedCode
149+ | Closed !ClosedCode
194150
195151isIdle :: StreamState -> Bool
196152isIdle Idle = True
@@ -217,79 +173,57 @@ instance Show StreamState where
217173----------------------------------------------------------------
218174
219175data Stream = Stream {
220− streamNumber :: StreamId
221− , streamState :: IORef StreamState
176+ streamNumber :: !StreamId
177+ , streamState :: !(IORef StreamState)
222178 -- Next two fields are for error checking.
223− , streamContentLength :: IORef (Maybe Int)
224− , streamBodyLength :: IORef Int
225− , streamWindow :: TVar WindowSize
226− , streamPrecedence :: IORef Precedence
227− -- | The concurrency IORef in which this stream has been counted. The client
228− -- and server each have separate concurrency values to respect, so pushed
229− -- streams need to decrement a different count when they're closed. This
230− -- should be either @concurrency ctx@ or @pushConcurrency ctx@.
231− , concurrencyRef :: IORef Int
179+ , streamContentLength :: !(IORef (Maybe Int))
180+ , streamBodyLength :: !(IORef Int)
181+ , streamWindow :: !(TVar WindowSize)
182+ , streamPrecedence :: !(IORef Precedence)
232183 }
233184
234185instance Show Stream where
235186 show s = show (streamNumber s)
236187
237−newStream :: IORef Int -> StreamId -> WindowSize -> IO Stream
238−newStream ref sid win =
239− Stream sid <$> newIORef Idle
240− <*> newIORef Nothing
241− <*> newIORef 0
242− <*> newTVarIO win
243− <*> newIORef defaultPrecedence
244− <*> pure ref
188+newStream :: StreamId -> WindowSize -> IO Stream
189+newStream sid win = Stream sid <$> newIORef Idle
190+ <*> newIORef Nothing
191+ <*> newIORef 0
192+ <*> newTVarIO win
193+ <*> newIORef defaultPrecedence
245194
246195----------------------------------------------------------------
247196
248−opened :: Stream -> IO ()
249−opened Stream{concurrencyRef,streamState} = do
250− atomicModifyIORef' concurrencyRef (\x -> (x+1,()))
197+opened :: Context -> Stream -> IO ()
198+opened Context{concurrency} Stream{streamState} = do
199+ atomicModifyIORef' concurrency (\x -> (x+1,()))
251200 writeIORef streamState (Open JustOpened)
252201
253−closed :: Stream -> ClosedCode -> IO ()
254−closed Stream{concurrencyRef,streamState} cc = do
255− atomicModifyIORef' concurrencyRef (\x -> (x-1,()))
256− writeIORef streamState (Closed cc)
202+closed :: Context -> Stream -> ClosedCode -> IO ()
203+closed Context{concurrency,streamTable} Stream{streamState,streamNumber} cc = do
204+ remove streamTable streamNumber
205+ atomicModifyIORef' concurrency (\x -> (x-1,()))
206+ writeIORef streamState (Closed cc) -- anyway
257207
258208----------------------------------------------------------------
259209
260−type StreamTable = Reaper (IntMap Stream) (M.Key, Stream)
210+newtype StreamTable = StreamTable (IORef (IntMap Stream))
261211
262−initialize :: Int -> IO StreamTable
263−initialize duration = mkReaper settings
264− where
265− settings = defaultReaperSettings {
266− reaperAction = clean
267− , reaperDelay = duration * 1000000
268− , reaperCons = uncurry M.insert
269− , reaperNull = M.null
270− , reaperEmpty = M.empty
271− }
272−
273−clean :: IntMap Stream -> IO (IntMap Stream -> IntMap Stream)
274−clean old = do
275− new <- M.fromAscList <$> prune oldlist []
276− return $ M.union new
277− where
278− oldlist = M.toDescList old
279− prune [] lst = return lst
280− prune (x@(_,s):xs) lst = do
281− st <- readIORef (streamState s)
282− if isClosed st then
283− prune xs lst
284− else
285− prune xs (x:lst)
212+newStreamTable :: IO StreamTable
213+newStreamTable = StreamTable <$> newIORef M.empty
286214
287215insert :: StreamTable -> M.Key -> Stream -> IO ()
288−insert strmtbl k v = reaperAdd strmtbl (k,v)
216+insert (StreamTable ref) k v = atomicModifyIORef' ref $ \m ->
217+ let !m' = M.insert k v m
218+ in (m', ())
289219
290−search :: StreamTable -> M.Key -> IO (Maybe Stream)
291−search strmtbl k = M.lookup k <$> reaperRead strmtbl
220+remove :: StreamTable -> M.Key -> IO ()
221+remove (StreamTable ref) k = atomicModifyIORef' ref $ \m ->
222+ let !m' = M.delete k m
223+ in (m', ())
292224
225+search :: StreamTable -> M.Key -> IO (Maybe Stream)
226+search (StreamTable ref) k = M.lookup k <$> readIORef ref
293227
294228-- INVARIANT: streams in the output queue have non-zero window size.
295229enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+126−173
@@ -1,11 +1,12 @@
1−{-# LANGUAGE DeriveDataTypeable #-}
2−{-# LANGUAGE RankNTypes #-}
31{-# LANGUAGE NamedFieldPuns #-}
42{-# LANGUAGE PatternGuards #-}
53{-# LANGUAGE CPP #-}
4+{-# LANGUAGE ScopedTypeVariables #-}
5+{-# LANGUAGE OverloadedStrings #-}
6+{-# LANGUAGE BangPatterns #-}
67
78module Network.Wai.Handler.Warp.HTTP2.Worker (
8− Respond
9+ Responder
910 , response
1011 , worker
1112 ) where
@@ -15,212 +16,164 @@ import Control.Applicative
1516#endif
1617import Control.Concurrent
1718import Control.Concurrent.STM
18−import Control.Exception (Exception, SomeException(..), AsyncException(..))
19+import Control.Exception (SomeException(..), AsyncException(..))
1920import qualified Control.Exception as E
2021import Control.Monad (void, when)
21−import Data.Typeable
22+import Data.ByteString.Builder (byteString)
2223import qualified Network.HTTP.Types as H
2324import Network.HTTP2
2425import Network.HTTP2.Priority
2526import Network.Wai
27+import Network.Wai.Handler.Warp.File
28+import Network.Wai.Handler.Warp.FileInfoCache
2629import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
2730import Network.Wai.Handler.Warp.HTTP2.Manager
2831import Network.Wai.Handler.Warp.HTTP2.Types
32+import Network.Wai.Handler.Warp.Header
2933import Network.Wai.Handler.Warp.IORef
30−import Network.Wai.HTTP2
31− ( Chunk(..)
32− , HTTP2Application
33− , PushPromise
34− , Responder(runResponder)
35− , RespondFunc
36− )
34+import qualified Network.Wai.Handler.Warp.Response as R
3735import qualified Network.Wai.Handler.Warp.Settings as S
3836import qualified Network.Wai.Handler.Warp.Timeout as T
37+import Network.Wai.Handler.Warp.Types
38+import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceived(..))
3939
4040----------------------------------------------------------------
4141
42−-- | An 'HTTP2Application' takes a function of status, headers, trailers, and
43−-- body; this type implements that by currying some internal arguments.
44−--
45−-- The token type of the RespondFunc is set to be (). This is a bit
46−-- anti-climactic, but the real benefit of the token type is that the
47−-- application is forced to call the responder, and making it a boring type
48−-- doesn't break that property.
49−--
50−-- This is the argument to a 'Responder'.
51−type Respond = IO () -> Stream -> RespondFunc ()
52−
53−-- | This function is passed to workers. They also pass responses from
54−-- 'HTTP2Application's to this function. This function enqueues commands for
55−-- the HTTP/2 sender.
56−response :: Context -> Manager -> ThreadContinue -> Respond
57−response ctx mgr tconf tickle strm s h strmbdy = do
58− -- TODO(awpr) HEAD requests will still stream.
59−
60− -- We must not exit this WAI application.
61− -- If the application exits, streaming would be also closed.
62− -- So, this work occupies this thread.
63− --
64− -- We need to increase the number of workers.
65− myThreadId >>= replaceWithAction mgr
66− -- After this work, this thread stops to decrease the number of workers.
67− setThreadContinue tconf False
68−
69− runStream ctx OResponse tickle strm s h strmbdy
70−
71−-- | Set up a waiter thread and run the stream body with functions to enqueue
72−-- 'Sequence's on the stream's queue.
73−runStream :: Context
74− -> (Stream -> H.Status -> H.ResponseHeaders -> Aux -> Output)
75− -> Respond
76−runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do
77− -- Since 'Body' is loop, we cannot control it.
78− -- So, let's serialize 'Builder' with a designated queue.
79− sq <- newTBQueueIO 10 -- fixme: hard coding: 10
80− tvar <- newTVarIO SyncNone
81− let out = mkOutput strm s h (Persist sq tvar)
82− -- Since we must not enqueue an empty queue to the priority
83− -- queue, we spawn a thread to ensure that the designated
84− -- queue is not empty.
85− void $ forkIO $ waiter tvar sq strm outputQ
86− atomically $ writeTVar tvar $ SyncNext out
87− let write chunk = do
88− atomically $ writeTBQueue sq $ case chunk of
89− BuilderChunk b -> SBuilder b
90− FileChunk path part -> SFile path part
91− tickle
92− flush = atomically $ writeTBQueue sq SFlush
93− trailers <- strmbdy write flush
94− atomically $ writeTBQueue sq $ SFinish trailers
95−
96−-- | Handle abnormal termination of a stream: mark it as closed, send a reset
97−-- frame, and call the user's 'settingsOnException' handler if applicable.
98−cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()
99−cleanupStream Context{outputQ} set strm req me = do
100− closed strm Killed
101− let sid = streamNumber strm
102− frame = resetFrame InternalError sid
103− enqueueControl outputQ sid $ OFrame frame
104− case me of
105− Nothing -> return ()
106− Just e -> S.settingsOnException set req e
107−
108−-- | Push the given 'Responder' to the client if the settings allow it
109−-- (specifically 'enablePush' and 'maxConcurrentStreams'). Returns 'True' if
110−-- the stream was actually pushed.
111−--
112−-- This is the push function given to an 'HTTP2Application'.
113−pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
114−pushResponder ctx set strm promise responder = do
115− let Context{ http2settings
116− , pushConcurrency
117− } = ctx
118− cnt <- readIORef pushConcurrency
119− settings <- readIORef http2settings
120− let enabled = enablePush settings
121− fits = maybe True (cnt <) $ maxConcurrentStreams settings
122− canPush = fits && enabled
123− if canPush then
124− actuallyPushResponder ctx set strm promise responder
125− else
126− return False
127−
128−-- | Set up a pushed stream and run the 'Responder' in its own thread. Waits
129−-- for the sender thread to handle the push request. This can fail to push the
130−-- stream and return 'False' if the sender dequeued the push request after the
131−-- associated stream was closed.
132−actuallyPushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool
133−actuallyPushResponder ctx set strm promise responder = do
134− let Context{ http2settings
135− , nextPushStreamId
136− , pushConcurrency
137− , streamTable
138− } = ctx
139− -- Claim the next outgoing stream.
140− newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)
141− ws <- initialWindowSize <$> readIORef http2settings
142−
143− newStrm <- newStream pushConcurrency newSid ws
144− -- Section 5.3.5 of RFC 7540 defines the weight of push promise is 16.
145− -- But we need not to follow the spec. So, this value would change
146− -- if necessary.
147− writeIORef (streamPrecedence newStrm) $
148− toPrecedence $ defaultPriority { streamDependency = streamNumber strm }
149− opened newStrm
150− insert streamTable newSid newStrm
151−
152− -- Set up a channel for the sender to report back whether it pushed the
153− -- stream.
154− mvar <- newEmptyMVar
155−
156− let mkOutput = OPush strm promise mvar
157− tickle = return ()
158− respond = runStream ctx mkOutput
159−
160− -- TODO(awpr): synthesize a Request for 'settingsOnException'?
161− _ <- forkIO $ runResponder responder (respond tickle newStrm) `E.catch`
162− (cleanupStream ctx set strm Nothing . Just)
163−
164− takeMVar mvar
165−
166−data Break = Break deriving (Show, Typeable)
167−
168−instance Exception Break
169−
170−worker :: Context
171− -> S.Settings
172− -> T.Manager
173− -> HTTP2Application
174− -> (ThreadContinue -> Respond)
175− -> IO ()
176−worker ctx@Context{inputQ} set tm app respond = do
177− tid <- myThreadId
42+-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.
43+-- This type implements the second argument (Response -> IO ResponseReceived)
44+-- with extra arguments.
45+type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
46+ Response -> IO ResponseReceived
47+
48+-- | This function is passed to workers.
49+-- They also pass 'Response's from 'Application's to this function.
50+-- This function enqueues commands for the HTTP/2 sender.
51+response :: InternalInfo -> S.Settings -> Context -> Manager -> Responder
52+response ii settings Context{outputQ} mgr tconf th strm req rsp
53+ | R.hasBody s0 = case rsp of
54+ ResponseStream _ _ strmbdy
55+ | isHead -> responseNoBody s0 hs0
56+ | otherwise -> responseStreaming strmbdy
57+ ResponseBuilder _ _ b
58+ | isHead -> responseNoBody s0 hs0
59+ | otherwise -> responseBuilderBody s0 hs0 b
60+ ResponseFile _ _ p mp -> responseFileXXX p mp
61+ ResponseRaw _ _ -> error "HTTP/2 does not support ResponseRaw"
62+ | otherwise = responseNoBody s0 hs0
63+ where
64+ !isHead = requestMethod req == H.methodHead
65+ !s0 = responseStatus rsp
66+ !hs0 = responseHeaders rsp
67+ !logger = S.settingsLogger settings req
68+
69+ responseNoBody s hs = responseBuilderBody s hs mempty
70+
71+ responseBuilderBody s hs bdy = do
72+ logger s Nothing
73+ setThreadContinue tconf True
74+ let rsp' = ResponseBuilder s hs bdy
75+ out = OResponse strm rsp' (Oneshot True)
76+ enqueueOrSpawnTemporaryWaiter strm outputQ out
77+ return ResponseReceived
78+
79+ responseFileXXX path Nothing = do
80+ efinfo <- E.try $ fileInfo ii path
81+ case efinfo of
82+ Left (_ex :: E.IOException) -> response404
83+ Right finfo -> case conditionalRequest finfo hs0 (indexRequestHeader (requestHeaders req)) of
84+ WithoutBody s -> responseNoBody s hs0
85+ WithBody s hs beg len -> responseFile2XX s hs path (Just (FilePart beg len (fileInfoSize finfo)))
86+
87+ responseFileXXX path mpart = responseFile2XX s0 hs0 path mpart
88+
89+ responseFile2XX s hs path mpart
90+ | isHead = responseNoBody s hs
91+ | otherwise = do
92+ logger s (filePartByteCount <$> mpart)
93+ setThreadContinue tconf True
94+ let rsp' = ResponseFile s hs path mpart
95+ out = OResponse strm rsp' (Oneshot True)
96+ enqueueOrSpawnTemporaryWaiter strm outputQ out
97+ return ResponseReceived
98+
99+ response404 = responseBuilderBody s hs body
100+ where
101+ s = H.notFound404
102+ hs = R.replaceHeader H.hContentType "text/plain; charset=utf-8" hs0
103+ body = byteString "File not found"
104+
105+ responseStreaming strmbdy = do
106+ logger s0 Nothing
107+ -- We must not exit this WAI application.
108+ -- If the application exits, streaming would be also closed.
109+ -- So, this work occupies this thread.
110+ --
111+ -- We need to increase the number of workers.
112+ myThreadId >>= replaceWithAction mgr
113+ -- After this work, this thread stops to decease
114+ -- the number of workers.
115+ setThreadContinue tconf False
116+ -- Since 'StreamingBody' is loop, we cannot control it.
117+ -- So, let's serialize 'Builder' with a designated queue.
118+ sq <- newTBQueueIO 10 -- fixme: hard coding: 10
119+ tvar <- newTVarIO SyncNone
120+ let out = OResponse strm rsp (Persist sq tvar)
121+ -- Since we must not enqueue an empty queue to the priority
122+ -- queue, we spawn a thread to ensure that the designated
123+ -- queue is not empty.
124+ void $ forkIO $ waiter tvar sq outputQ
125+ atomically $ writeTVar tvar (SyncNext out)
126+ let push b = do
127+ atomically $ writeTBQueue sq (SBuilder b)
128+ T.tickle th
129+ flush = atomically $ writeTBQueue sq SFlush
130+ _ <- strmbdy push flush
131+ atomically $ writeTBQueue sq SFinish
132+ return ResponseReceived
133+
134+worker :: Context -> S.Settings -> Application -> Responder -> T.Manager -> IO ()
135+worker ctx@Context{inputQ,outputQ} set app responder tm = do
178136 sinfo <- newStreamInfo
179137 tcont <- newThreadContinue
180− let setup = T.register tm $ E.throwTo tid Break
181− E.bracket setup T.cancel $ go sinfo tcont
138+ E.bracket (T.registerKillThread tm) T.cancel $ go sinfo tcont
182139 where
183140 go sinfo tcont th = do
184141 setThreadContinue tcont True
185−
186142 ex <- E.try $ do
187143 T.pause th
188144 Input strm req <- atomically $ readTQueue inputQ
189145 setStreamInfo sinfo strm req
190146 T.resume th
191147 T.tickle th
192− let responder = app req $ pushResponder ctx set strm
193− runResponder responder $ respond tcont (T.tickle th) strm
148+ app req $ responder tcont th strm req
194149 cont1 <- case ex of
195− Right () -> return True
150+ Right ResponseReceived -> return True
196151 Left e@(SomeException _)
197− | Just Break <- E.fromException e -> do
152+ -- killed by the worker manager
153+ | Just ThreadKilled <- E.fromException e -> return False
154+ | Just T.TimeoutThread <- E.fromException e -> do
198155 cleanup sinfo Nothing
199156 return True
200− -- killed by the sender
201− | Just ThreadKilled <- E.fromException e -> do
202− cleanup sinfo Nothing
203− return False
204157 | otherwise -> do
205− cleanup sinfo (Just e)
158+ cleanup sinfo $ Just e
206159 return True
207160 cont2 <- getThreadContinue tcont
161+ clearStreamInfo sinfo
208162 when (cont1 && cont2) $ go sinfo tcont th
209163 cleanup sinfo me = do
210164 m <- getStreamInfo sinfo
211165 case m of
212166 Nothing -> return ()
213167 Just (strm,req) -> do
214− cleanupStream ctx set strm (Just req) me
215− clearStreamInfo sinfo
216−
217−-- | A dedicated waiter thread to re-enqueue the stream in the priority tree
218−-- whenever output becomes available. When the sender drains the queue and
219−-- moves on to another stream, it drops a message in the 'TVar', and this
220−-- thread wakes up, waits for more output to become available, and re-enqueues
221−-- the stream.
222−waiter :: TVar Sync -> TBQueue Sequence -> Stream -> PriorityTree Output -> IO ()
223−waiter tvar sq strm outQ = do
168+ closed ctx strm Killed
169+ let frame = resetFrame InternalError (streamNumber strm)
170+ enqueueControl outputQ 0 (OFrame frame)
171+ case me of
172+ Nothing -> return ()
173+ Just e -> S.settingsOnException set (Just req) e
174+
175+waiter :: TVar Sync -> TBQueue Sequence -> PriorityTree Output -> IO ()
176+waiter tvar sq outQ = do
224177 -- waiting for actions other than SyncNone
225178 mx <- atomically $ do
226179 mout <- readTVar tvar
@@ -231,7 +184,7 @@ waiter tvar sq strm outQ = do
231184 return $ Just out
232185 SyncFinish -> return Nothing
233186 case mx of
234− Nothing -> return ()
187+ Nothing -> return ()
235188 Just out -> do
236189 -- ensuring that the streaming queue is not empty.
237190 atomically $ do
@@ -239,14 +192,14 @@ waiter tvar sq strm outQ = do
239192 when isEmpty retry
240193 -- ensuring that stream window is greater than 0.
241194 enqueueWhenWindowIsOpen outQ out
242− waiter tvar sq strm outQ
195+ waiter tvar sq outQ
243196
244197----------------------------------------------------------------
245198
246199-- | It would nice if responders could return values to workers.
247200-- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0.
248201-- It is not wise to change this type.
249−-- So, a reference is shared by a 'Respond' and its worker.
202+-- So, a reference is shared by a responder and its worker.
250203-- The reference refers a value of this type as a return value.
251204-- If 'True', the worker continue to serve requests.
252205-- Otherwise, the worker get finished.
warp/Network/Wai/Handler/Warp/Internal.hs−11
@@ -8,18 +8,7 @@ module Network.Wai.Handler.Warp.Internal (
88 , runSettingsConnection
99 , runSettingsConnectionMaker
1010 , runSettingsConnectionMakerSecure
11− , runServe
12− , runServeEnv
13− , runServeSettings
14− , runServeSettingsSocket
15− , runServeSettingsConnection
16− , runServeSettingsConnectionMaker
17− , runServeSettingsConnectionMakerSecure
1811 , Transport (..)
19− -- * ServeConnection
20− , ServeConnection
21− , serveDefault
22− , serveHTTP2
2312 -- * Connection
2413 , Connection (..)
2514 , socketConnection
warp/Network/Wai/Handler/Warp/RequestHeader.hs+1−2
@@ -3,7 +3,6 @@
33
44module Network.Wai.Handler.Warp.RequestHeader (
55 parseHeaderLines
6− , parseByteRanges
76 ) where
87
98import Control.Exception (throwIO)
@@ -18,7 +17,7 @@ import Foreign.Ptr (Ptr, plusPtr, minusPtr, nullPtr)
1817import Foreign.Storable (peek)
1918import qualified Network.HTTP.Types as H
2019import Network.Wai.Handler.Warp.Types
21−import Network.Wai.Internal (parseByteRanges)
20+
2221-- $setup
2322-- >>> :set -XOverloadedStrings
2423
warp/Network/Wai/Handler/Warp/Response.hs+1
@@ -11,6 +11,7 @@ module Network.Wai.Handler.Warp.Response (
1111 , addDate
1212 , addServer
1313 , hasBody
14+ , replaceHeader
1415 ) where
1516
1617#ifndef MIN_VERSION_base
warp/Network/Wai/Handler/Warp/Run.hs+28−104
@@ -25,7 +25,6 @@ import Network (sClose, Socket)
2525import Network.Socket (accept, withSocketsDo, SockAddr(SockAddrInet, SockAddrInet6))
2626import qualified Network.Socket.ByteString as Sock
2727import Network.Wai
28−import Network.Wai.HTTP2 (HTTP2Application, promoteApplication)
2928import Network.Wai.Handler.Warp.Buffer
3029import Network.Wai.Handler.Warp.Counter
3130import qualified Network.Wai.Handler.Warp.Date as D
@@ -74,23 +73,10 @@ allowInterrupt :: IO ()
7473allowInterrupt = unblock $ return ()
7574#endif
7675
77−-- Composition over two arguments at once; used for runHTTP2\*.
78−(.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d
79−f .: g = curry $ f . uncurry g
80−
8176-- | Run an 'Application' on the given port.
8277-- This calls 'runSettings' with 'defaultSettings'.
8378run :: Port -> Application -> IO ()
84−run port = runServe port . serveDefault
85−
86−-- | Serve an 'HTTP2Application' and an 'Application' together on the given
87−-- port.
88−runHTTP2 :: Port -> HTTP2Application -> Application -> IO ()
89−runHTTP2 port = runServe port .: serveHTTP2
90−
91−-- | The generalized form of 'run'.
92−runServe :: Port -> ServeConnection -> IO ()
93−runServe p = runServeSettings defaultSettings { settingsPort = p }
79+run p = runSettings defaultSettings { settingsPort = p }
9480
9581-- | Run an 'Application' on the port present in the @PORT@
9682-- environment variable. Uses the 'Port' given when the variable is unset.
@@ -98,44 +84,28 @@ runServe p = runServeSettings defaultSettings { settingsPort = p }
9884--
9985-- Since 3.0.9
10086runEnv :: Port -> Application -> IO ()
101−runEnv port = runServeEnv port . serveDefault
102−
103−-- | The HTTP\/2-aware form of 'runEnv'.
104−runHTTP2Env :: Port -> HTTP2Application -> Application -> IO ()
105−runHTTP2Env port = runServeEnv port .: serveHTTP2
106−
107−-- | The generalized form of 'runEnv'.
108−runServeEnv :: Port -> ServeConnection -> IO ()
109−runServeEnv p serveConn = do
87+runEnv p app = do
11088 mp <- lookup "PORT" <$> getEnvironment
11189
112− maybe (runServe p serveConn) runReadPort mp
90+ maybe (run p app) runReadPort mp
11391
11492 where
11593 runReadPort :: String -> IO ()
11694 runReadPort sp = case reads sp of
117− ((p', _):_) -> runServe p' serveConn
95+ ((p', _):_) -> run p' app
11896 _ -> fail $ "Invalid value in $PORT: " ++ sp
11997
12098-- | Run an 'Application' with the given 'Settings'.
12199-- This opens a listen socket on the port defined in 'Settings' and
122100-- calls 'runSettingsSocket'.
123101runSettings :: Settings -> Application -> IO ()
124−runSettings set = runServeSettings set . serveDefault
125−
126−-- | The HTTP\/2-aware form of 'runSettings'.
127−runHTTP2Settings :: Settings -> HTTP2Application -> Application -> IO ()
128−runHTTP2Settings set = runServeSettings set .: serveHTTP2
129−
130−-- | The generalized form of 'runSettings'.
131−runServeSettings :: Settings -> ServeConnection -> IO ()
132−runServeSettings set serveConn = withSocketsDo $
102+runSettings set app = withSocketsDo $
133103 bracket
134104 (bindPortTCP (settingsPort set) (settingsHost set))
135105 sClose
136106 (\socket -> do
137107 setSocketCloseOnExec socket
138− runServeSettingsSocket set socket serveConn)
108+ runSettingsSocket set socket app)
139109
140110-- | This installs a shutdown handler for the given socket and
141111-- calls 'runSettingsConnection' with the default connection setup action
@@ -149,22 +119,9 @@ runServeSettings set serveConn = withSocketsDo $
149119-- Note that the 'settingsPort' will still be passed to 'Application's via the
150120-- 'serverPort' record.
151121runSettingsSocket :: Settings -> Socket -> Application -> IO ()
152−runSettingsSocket set socket = runServeSettingsSocket set socket . serveDefault
153−
154−-- | The HTTP\/2-aware form of 'runSettingsSocket'.
155−runHTTP2SettingsSocket :: Settings
156− -> Socket
157− -> HTTP2Application
158− -> Application
159− -> IO ()
160−runHTTP2SettingsSocket set socket =
161− runServeSettingsSocket set socket .: serveHTTP2
162−
163−-- | The generalized form of 'runSettingsSocket'.
164−runServeSettingsSocket :: Settings -> Socket -> ServeConnection -> IO ()
165−runServeSettingsSocket set socket serveConn = do
122+runSettingsSocket set socket app = do
166123 settingsInstallShutdownHandler set closeListenSocket
167− runServeSettingsConnection set getConn serveConn
124+ runSettingsConnection set getConn app
168125 where
169126 getConn = do
170127#if WINDOWS
@@ -188,16 +145,7 @@ runServeSettingsSocket set socket serveConn = do
188145--
189146-- Since 1.3.5
190147runSettingsConnection :: Settings -> IO (Connection, SockAddr) -> Application -> IO ()
191−runSettingsConnection set getConn =
192− runServeSettingsConnection set getConn . serveDefault
193−
194−-- | The generalized form of 'runSettingsConnection'.
195−runServeSettingsConnection :: Settings
196− -> IO (Connection, SockAddr)
197− -> ServeConnection
198− -> IO ()
199−runServeSettingsConnection set getConn serveConn =
200− runServeSettingsConnectionMaker set getConnMaker serveConn
148+runSettingsConnection set getConn app = runSettingsConnectionMaker set getConnMaker app
201149 where
202150 getConnMaker = do
203151 (conn, sa) <- getConn
@@ -206,16 +154,8 @@ runServeSettingsConnection set getConn serveConn =
206154-- | This modifies the connection maker so that it returns 'TCP' for 'Transport'
207155-- (i.e. plain HTTP) then calls 'runSettingsConnectionMakerSecure'.
208156runSettingsConnectionMaker :: Settings -> IO (IO Connection, SockAddr) -> Application -> IO ()
209−runSettingsConnectionMaker set getConnMaker =
210− runServeSettingsConnectionMaker set getConnMaker . serveDefault
211−
212−-- | The generalized form of 'runSettingsConnectionMaker'.
213−runServeSettingsConnectionMaker :: Settings
214− -> IO (IO Connection, SockAddr)
215− -> ServeConnection
216− -> IO ()
217−runServeSettingsConnectionMaker x y =
218− runServeSettingsConnectionMakerSecure x (toTCP <$> y)
157+runSettingsConnectionMaker x y =
158+ runSettingsConnectionMakerSecure x (toTCP <$> y)
219159 where
220160 toTCP = first ((, TCP) <$>)
221161
@@ -228,18 +168,10 @@ runServeSettingsConnectionMaker x y =
228168--
229169-- Since 2.1.4
230170runSettingsConnectionMakerSecure :: Settings -> IO (IO (Connection, Transport), SockAddr) -> Application -> IO ()
231−runSettingsConnectionMakerSecure set getConnMaker =
232− runServeSettingsConnectionMakerSecure set getConnMaker . serveDefault
233−
234−-- | The generalized form of 'runSettingsConnectionMakerSecure'.
235−runServeSettingsConnectionMakerSecure :: Settings
236− -> IO (IO (Connection, Transport), SockAddr)
237− -> ServeConnection
238− -> IO ()
239−runServeSettingsConnectionMakerSecure set getConnMaker serveConn = do
171+runSettingsConnectionMakerSecure set getConnMaker app = do
240172 settingsBeforeMainLoop set
241173 counter <- newCounter
242− withII $ acceptConnection set getConnMaker serveConn counter
174+ withII $ acceptConnection set getConnMaker app counter
243175 where
244176 withII action =
245177 D.withDateCache $ \dc ->
@@ -273,11 +205,11 @@ runServeSettingsConnectionMakerSecure set getConnMaker serveConn = do
273205-- Our approach is explained in the comments below.
274206acceptConnection :: Settings
275207 -> IO (IO (Connection, Transport), SockAddr)
276− -> ServeConnection
208+ -> Application
277209 -> Counter
278210 -> InternalInfo
279211 -> IO ()
280−acceptConnection set getConnMaker serveConn counter ii0 = do
212+acceptConnection set getConnMaker app counter ii0 = do
281213 -- First mask all exceptions in acceptLoop. This is necessary to
282214 -- ensure that no async exception is throw between the call to
283215 -- acceptNewConnection and the registering of connClose.
@@ -299,7 +231,7 @@ acceptConnection set getConnMaker serveConn counter ii0 = do
299231 case mx of
300232 Nothing -> return ()
301233 Just (mkConn, addr) -> do
302− fork set mkConn addr serveConn counter ii0
234+ fork set mkConn addr app counter ii0
303235 acceptLoop
304236
305237 acceptNewConnection = do
@@ -323,11 +255,11 @@ acceptConnection set getConnMaker serveConn counter ii0 = do
323255fork :: Settings
324256 -> IO (Connection, Transport)
325257 -> SockAddr
326− -> ServeConnection
258+ -> Application
327259 -> Counter
328260 -> InternalInfo
329261 -> IO ()
330−fork set mkConn addr serveConn counter ii0 = settingsFork set $ \ unmask ->
262+fork set mkConn addr app counter ii0 = settingsFork set $ \ unmask ->
331263 -- Run the connection maker to get a new connection, and ensure
332264 -- that the connection is closed. If the mkConn call throws an
333265 -- exception, we will leak the connection. If the mkConn call is
@@ -358,29 +290,21 @@ fork set mkConn addr serveConn counter ii0 = settingsFork set $ \ unmask ->
358290
359291 -- Actually serve this connection.
360292 -- bracket with closeConn above ensures the connection is closed.
361− when goingon $ serveConn conn ii addr transport set
293+ when goingon $ serveConnection conn ii addr transport set app
362294 where
363295 closeConn (conn, _transport) = connClose conn
364296
365297 onOpen adr = increase counter >> settingsOnOpen set adr
366298 onClose adr _ = decrease counter >> settingsOnClose set adr
367299
368−-- The type of a function to serve a fully-prepared connection.
369−type ServeConnection = Connection
370− -> InternalInfo
371− -> SockAddr
372− -> Transport
373− -> Settings
374− -> IO ()
375−
376−-- Serve an HTTP\/2-unaware Application to a connection over any HTTP version.
377−serveDefault :: Application -> ServeConnection
378−serveDefault app = serveHTTP2 (promoteApplication app) app
379−
380−-- Serve an HTTP\/2-aware application over HTTP\/2 or a backup 'Application'
381−-- over HTTP\/1.1 or HTTP\/1.
382−serveHTTP2 :: HTTP2Application -> Application -> ServeConnection
383−serveHTTP2 app2 app conn ii origAddr transport settings = do
300+serveConnection :: Connection
301+ -> InternalInfo
302+ -> SockAddr
303+ -> Transport
304+ -> Settings
305+ -> Application
306+ -> IO ()
307+serveConnection conn ii origAddr transport settings app = do
384308 -- fixme: Upgrading to HTTP/2 should be supported.
385309 (h2,bs) <- if isHTTP2 transport then
386310 return (True, "")
@@ -393,7 +317,7 @@ serveHTTP2 app2 app conn ii origAddr transport settings = do
393317 if settingsHTTP2Enabled settings && h2 then do
394318 recvN <- makeReceiveN bs (connRecv conn) (connRecvBuf conn)
395319 -- fixme: origAddr
396− http2 conn ii origAddr transport settings recvN app2
320+ http2 conn ii origAddr transport settings recvN app
397321 else do
398322 istatus <- newIORef False
399323 src <- mkSource (wrappedRecv conn th istatus (settingsSlowlorisSize settings))
warp/test/RequestSpec.hs+1−1
@@ -4,8 +4,8 @@
44
55module RequestSpec (main, spec) where
66
7+import Network.Wai.Handler.Warp.File (parseByteRanges)
78import Network.Wai.Handler.Warp.Request
8−import Network.Wai.Handler.Warp.RequestHeader (parseByteRanges)
99import Network.Wai.Handler.Warp.Types
1010import Test.Hspec
1111import Test.Hspec.QuickCheck