diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,8 @@ +## 3.1.0 + +* Supporting HTTP/2 [#399](https://github.com/yesodweb/wai/pull/399) +* Cleaning up APIs [#387](https://github.com/yesodweb/wai/issues/387) + ## 3.0.13.1 * Remove dependency on the void package [#375](https://github.com/yesodweb/wai/pull/375) diff --git a/Network/Wai/Handler/Warp.hs b/Network/Wai/Handler/Warp.hs --- a/Network/Wai/Handler/Warp.hs +++ b/Network/Wai/Handler/Warp.hs @@ -1,5 +1,6 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE PatternGuards #-} {-# OPTIONS_GHC -fno-warn-deprecations #-} --------------------------------------------------------- @@ -18,6 +19,9 @@ -- | A fast, light-weight HTTP server handler for WAI. -- +-- HTTP\/1.0, HTTP\/1.1 and HTTP\/2 are supported. For HTTP\/2, +-- Warp supports direct and ALPN (in TLS) but not upgrade. +-- -- Note on slowloris timeouts: to prevent slowloris attacks, timeouts are used -- at various points in request receiving and response sending. One interesting -- corner case is partial request body consumption; in that case, Warp's @@ -38,9 +42,6 @@ , runEnv , runSettings , runSettingsSocket - , runSettingsConnection - , runSettingsConnectionMaker - , runSettingsConnectionMakerSecure -- * Settings , Settings , defaultSettings @@ -66,34 +67,43 @@ -- ** Getters , getPort , getHost - -- ** Accessors - -- | Note: these accessors are deprecated, please use the @set@ versions instead. - , settingsPort - , settingsHost - , settingsOnException - , settingsOnExceptionResponse - , settingsOnOpen - , settingsOnClose - , settingsTimeout - , settingsManager - , settingsFdCacheDuration - , settingsBeforeMainLoop - , settingsNoParsePath - -- ** Debugging - , exceptionResponseForDebug + , getOnOpen + , getOnClose + , getOnException + -- ** Exception handler + , defaultOnException , defaultShouldDisplayException + -- ** Exception response handler + , defaultOnExceptionResponse + , exceptionResponseForDebug -- * Data types - , Transport (..) , HostPreference (..) , Port , InvalidRequest (..) - , ConnSendFileOverride (..) - -- * Per-request utilities + -- * Utilities , pauseTimeout - -- * Connection + -- * Internal + -- | The following APIs will be removed in Warp 3.2.0. Please use ones exported from Network.Wai.Handler.Warp.Internal. + + -- ** Low level run functions + , runSettingsConnection + , runSettingsConnectionMaker + , runSettingsConnectionMakerSecure + , Transport (..) + -- ** Connection , Connection (..) , socketConnection - -- * Internal + -- ** Buffer + , Buffer + , BufSize + , bufferSize + , allocateBuffer + , freeBuffer + -- ** Sendfile + , FileId (..) + , SendFile + , sendFile + , readSendFile -- ** Version , warpVersion -- ** Data types @@ -101,38 +111,41 @@ , HeaderValue , IndexedHeader , requestMaxIndex - -- ** Time out manager - , module Network.Wai.Handler.Warp.Timeout -- ** File descriptor cache , module Network.Wai.Handler.Warp.FdCache -- ** Date , module Network.Wai.Handler.Warp.Date -- ** Request and response + , Source , recvRequest , sendResponse + -- ** Time out manager + , module Network.Wai.Handler.Warp.Timeout ) where +import Control.Exception (SomeException) +import Data.ByteString (ByteString) +import Data.Maybe (fromMaybe) +import Data.Streaming.Network (HostPreference) +import qualified Data.Vault.Lazy as Vault +import Network.Socket (SockAddr) +import Network.Wai (Request, Response, vault) +import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.Date import Network.Wai.Handler.Warp.FdCache import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.Request import Network.Wai.Handler.Warp.Response import Network.Wai.Handler.Warp.Run +import Network.Wai.Handler.Warp.SendFile import Network.Wai.Handler.Warp.Settings import Network.Wai.Handler.Warp.Timeout import Network.Wai.Handler.Warp.Types -import Control.Exception (SomeException) -import Network.Wai (Request, Response, vault) -import Network.Socket (SockAddr) -import Data.Streaming.Network (HostPreference) -import Data.ByteString (ByteString) -import qualified Data.Vault.Lazy as Vault -import Data.Maybe (fromMaybe) -- | Port to listen on. Default value: 3000 -- -- Since 2.1.0 -setPort :: Int -> Settings -> Settings +setPort :: Port -> Settings -> Settings setPort x y = y { settingsPort = x } -- | Interface to bind to. Default value: HostIPv4 @@ -142,8 +155,7 @@ setHost x y = y { settingsHost = x } -- | What to do with exceptions thrown by either the application or server. --- Default: ignore server-generated exceptions (see 'InvalidRequest') and print --- application-generated exceptions to stderr. +-- Default: 'defaultOnException' -- -- Since 2.1.0 setOnException :: (Maybe Request -> SomeException -> IO ()) -> Settings -> Settings @@ -151,7 +163,7 @@ -- | A function to create a `Response` when an exception occurs. -- --- Default: 500, text/plain, \"Something went wrong\" +-- Default: 'defaultOnExceptionResponse' -- -- Since 2.1.0 setOnExceptionResponse :: (SomeException -> Response) -> Settings -> Settings @@ -221,7 +233,7 @@ -- | Get the listening port. -- -- Since 2.1.1 -getPort :: Settings -> Int +getPort :: Settings -> Port getPort = settingsPort -- | Get the interface to bind to. @@ -229,6 +241,18 @@ -- Since 2.1.1 getHost :: Settings -> HostPreference getHost = settingsHost + +-- | Get the action on opening connection. +getOnOpen :: Settings -> SockAddr -> IO Bool +getOnOpen = settingsOnOpen + +-- | Get the action on closeing connection. +getOnClose :: Settings -> SockAddr -> IO () +getOnClose = settingsOnClose + +-- | Get the exception handler. +getOnException :: Settings -> Maybe Request -> SomeException -> IO () +getOnException = settingsOnException -- | A code to install shutdown handler. -- diff --git a/Network/Wai/Handler/Warp/Buffer.hs b/Network/Wai/Handler/Warp/Buffer.hs --- a/Network/Wai/Handler/Warp/Buffer.hs +++ b/Network/Wai/Handler/Warp/Buffer.hs @@ -1,25 +1,108 @@ -module Network.Wai.Handler.Warp.Buffer where +{-# LANGUAGE BangPatterns, OverloadedStrings #-} +module Network.Wai.Handler.Warp.Buffer ( + bufferSize + , allocateBuffer + , freeBuffer + , mallocBS + , newBufferPool + , withBufferPool + , toBlazeBuffer + , copy + , bufferIO + ) where + +import Control.Monad (when) +import qualified Data.ByteString as BS +import Data.ByteString.Internal (ByteString(..), memcpy) +import Data.ByteString.Unsafe (unsafeTake, unsafeDrop) +import Data.IORef (newIORef, readIORef, writeIORef) import qualified Data.Streaming.ByteString.Builder.Buffer as B (Buffer (..)) -import Data.Word (Word8) -import Foreign.ForeignPtr (newForeignPtr_) -import Foreign.Marshal.Alloc (mallocBytes, free) -import Foreign.Ptr (Ptr, plusPtr) +import Foreign.ForeignPtr +import Foreign.Marshal.Alloc (mallocBytes, free, finalizerFree) +import Foreign.Ptr (castPtr, plusPtr) +import Network.Wai.Handler.Warp.Types -type Buffer = Ptr Word8 -type BufSize = Int +---------------------------------------------------------------- --- FIXME come up with good values here +-- | The default size of the write buffer: 16384 (2^14 = 1024 * 16). +-- This is the maximum size of TLS record. +-- This is also the maximum size of HTTP/2 frame payload +-- (excluding frame header). bufferSize :: BufSize -bufferSize = 4096 +bufferSize = 16384 +-- | Allocating a buffer with malloc(). allocateBuffer :: Int -> IO Buffer allocateBuffer = mallocBytes +-- | Releasing a buffer with free(). freeBuffer :: Buffer -> IO () freeBuffer = free +---------------------------------------------------------------- + +largeBufferSize :: Int +largeBufferSize = 16384 + +minBufferSize :: Int +minBufferSize = 2048 + +newBufferPool :: IO BufferPool +newBufferPool = newIORef BS.empty + +mallocBS :: Int -> IO ByteString +mallocBS size = do + ptr <- allocateBuffer size + fptr <- newForeignPtr finalizerFree ptr + return $! PS fptr 0 size +{-# INLINE mallocBS #-} + +usefulBuffer :: ByteString -> Bool +usefulBuffer buffer = BS.length buffer >= minBufferSize +{-# INLINE usefulBuffer #-} + +getBuffer :: BufferPool -> IO ByteString +getBuffer pool = do + buffer <- readIORef pool + if usefulBuffer buffer then return buffer else mallocBS largeBufferSize +{-# INLINE getBuffer #-} + +putBuffer :: BufferPool -> ByteString -> IO () +putBuffer pool buffer = when (usefulBuffer buffer) $ writeIORef pool buffer +{-# INLINE putBuffer #-} + +withForeignBuffer :: ByteString -> ((Buffer, BufSize) -> IO Int) -> IO Int +withForeignBuffer (PS ps s l) f = withForeignPtr ps $ \p -> f (castPtr p `plusPtr` s, l) +{-# INLINE withForeignBuffer #-} + +withBufferPool :: BufferPool -> ((Buffer, BufSize) -> IO Int) -> IO ByteString +withBufferPool pool f = do + buffer <- getBuffer pool + consumed <- withForeignBuffer buffer f + putBuffer pool $! unsafeDrop consumed buffer + return $! unsafeTake consumed buffer +{-# INLINE withBufferPool #-} + +---------------------------------------------------------------- +-- +-- Utilities +-- + toBlazeBuffer :: Buffer -> BufSize -> IO B.Buffer toBlazeBuffer ptr size = do fptr <- newForeignPtr_ ptr return $ B.Buffer fptr ptr ptr (ptr `plusPtr` size) + +-- | Copying the bytestring to the buffer. +-- This function returns the point where the next copy should start. +copy :: Buffer -> ByteString -> IO Buffer +copy !ptr (PS fp o l) = withForeignPtr fp $ \p -> do + memcpy ptr (p `plusPtr` o) (fromIntegral l) + return $! ptr `plusPtr` l +{-# INLINE copy #-} + +bufferIO :: Buffer -> Int -> (ByteString -> IO ()) -> IO () +bufferIO ptr siz io = do + fptr <- newForeignPtr_ ptr + io $ PS fptr 0 siz diff --git a/Network/Wai/Handler/Warp/Conduit.hs b/Network/Wai/Handler/Warp/Conduit.hs --- a/Network/Wai/Handler/Warp/Conduit.hs +++ b/Network/Wai/Handler/Warp/Conduit.hs @@ -1,13 +1,16 @@ +{-# LANGUAGE CPP #-} + module Network.Wai.Handler.Warp.Conduit where import Control.Exception import Control.Monad (when, unless) import Data.ByteString (ByteString) -import Data.ByteString.Lazy.Char8 (pack) import qualified Data.ByteString as S -import qualified Data.ByteString.Lazy as L import qualified Data.IORef as I -import Data.Word (Word, Word8) +#if __GLASGOW_HASKELL__ < 709 +import Data.Word (Word) +#endif +import Data.Word (Word8) import Network.Wai.Handler.Warp.Types ---------------------------------------------------------------- @@ -76,9 +79,6 @@ | DoneChunking deriving Show -bsCRLF :: L.ByteString -bsCRLF = pack "\r\n" - mkCSource :: Source -> IO CSource mkCSource src = do ref <- I.newIORef NeedLen @@ -148,13 +148,13 @@ return S.empty else do (x, y) <- - case S.breakByte 10 bs of + case S.break (== 10) bs of (x, y) | S.null y -> do bs2 <- readSource' src return $ if S.null bs2 then (x, y) - else S.breakByte 10 $ bs `S.append` bs2 + else S.break (== 10) $ bs `S.append` bs2 | otherwise -> return (x, y) let w = S.foldl' (\i c -> i * 16 + fromIntegral (hexToWord c)) 0 diff --git a/Network/Wai/Handler/Warp/Counter.hs b/Network/Wai/Handler/Warp/Counter.hs --- a/Network/Wai/Handler/Warp/Counter.hs +++ b/Network/Wai/Handler/Warp/Counter.hs @@ -1,24 +1,31 @@ +{-# LANGUAGE CPP #-} + module Network.Wai.Handler.Warp.Counter ( Counter , newCounter - , isZero + , waitForZero , increase , decrease ) where -import Network.Wai.Handler.Warp.IORef +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative ((<$>)) +#endif +import Control.Concurrent.STM +import Control.Monad (unless) -newtype Counter = Counter (IORef Int) +newtype Counter = Counter (TVar Int) newCounter :: IO Counter -newCounter = Counter <$> newIORef 0 +newCounter = Counter <$> newTVarIO 0 -isZero :: Counter -> IO Bool -isZero (Counter ref) = (== 0) <$> readIORef ref +waitForZero :: Counter -> IO () +waitForZero (Counter ref) = atomically $ do + x <- readTVar ref + unless (x == 0) retry increase :: Counter -> IO () -increase (Counter ref) = atomicModifyIORef' ref $ \x -> (x + 1, ()) +increase (Counter ref) = atomically $ modifyTVar' ref $ \x -> x + 1 decrease :: Counter -> IO () -decrease (Counter ref) = atomicModifyIORef' ref $ \x -> (x - 1, ()) +decrease (Counter ref) = atomically $ modifyTVar' ref $ \x -> x - 1 diff --git a/Network/Wai/Handler/Warp/Date.hs b/Network/Wai/Handler/Warp/Date.hs --- a/Network/Wai/Handler/Warp/Date.hs +++ b/Network/Wai/Handler/Warp/Date.hs @@ -7,7 +7,9 @@ , GMTDate ) where +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative +#endif import Control.AutoUpdate (defaultUpdateSettings, updateAction, mkAutoUpdate) import Data.ByteString.Char8 diff --git a/Network/Wai/Handler/Warp/FdCache.hs b/Network/Wai/Handler/Warp/FdCache.hs --- a/Network/Wai/Handler/Warp/FdCache.hs +++ b/Network/Wai/Handler/Warp/FdCache.hs @@ -2,7 +2,7 @@ -- | File descriptor cache to avoid locks in kernel. -#ifndef SENDFILEFD +#ifdef WINDOWS module Network.Wai.Handler.Warp.FdCache ( withFdCache , MutableFdCache @@ -22,7 +22,9 @@ , Refresh ) where +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative ((<$>), (<*>)) +#endif import Control.Exception (bracket) import Data.Hashable (hash) import Network.Wai.Handler.Warp.IORef diff --git a/Network/Wai/Handler/Warp/HTTP2.hs b/Network/Wai/Handler/Warp/HTTP2.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2.hs @@ -0,0 +1,66 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.Wai.Handler.Warp.HTTP2 (isHTTP2, http2) where + +import Control.Concurrent (forkIO, killThread) +import qualified Control.Exception as E +import Control.Monad (when, unless, replicateM_) +import Data.ByteString (ByteString) +import Network.HTTP2 +import Network.Socket (SockAddr) +import Network.Wai +import Network.Wai.Handler.Warp.HTTP2.EncodeFrame +import Network.Wai.Handler.Warp.HTTP2.Manager +import Network.Wai.Handler.Warp.HTTP2.Receiver +import Network.Wai.Handler.Warp.HTTP2.Request +import Network.Wai.Handler.Warp.HTTP2.Sender +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.HTTP2.Worker +import qualified Network.Wai.Handler.Warp.Settings as S (Settings) +import Network.Wai.Handler.Warp.Types + +---------------------------------------------------------------- + +http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> Application -> IO () +http2 conn ii addr transport settings readN app = do + checkTLS + ok <- checkPreface + when ok $ do + ctx <- newContext + -- Workers & Manager + mgr <- start + let responder = response ctx mgr + action = worker ctx settings tm app responder + setAction mgr action + -- fixme: hard coding: 10 + replicateM_ 10 $ spawnAction mgr + -- Receiver + let mkreq = mkRequest settings addr + tid <- forkIO $ frameReceiver ctx mkreq readN + -- Sender + -- frameSender is the main thread because it ensures to send + -- a goway frame. + frameSender ctx conn ii settings `E.finally` do + clearContext ctx + stop mgr + killThread tid + where + tm = timeoutManager ii + checkTLS = case transport of + TCP -> return () -- direct + tls -> unless (tls12orLater tls) $ goaway conn InadequateSecurity "Weak TLS" + tls12orLater tls = tlsMajorVersion tls == 3 && tlsMinorVersion tls >= 3 + checkPreface = do + preface <- readN connectionPrefaceLength + if connectionPreface /= preface then do + goaway conn ProtocolError "Preface mismatch" + return False + else + return True + +-- connClose must not be called here since Run:fork calls it +goaway :: Connection -> ErrorCodeId -> ByteString -> IO () +goaway Connection{..} etype debugmsg = connSendAll bytestream + where + bytestream = goawayFrame 0 etype debugmsg diff --git a/Network/Wai/Handler/Warp/HTTP2/EncodeFrame.hs b/Network/Wai/Handler/Warp/HTTP2/EncodeFrame.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/EncodeFrame.hs @@ -0,0 +1,35 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Network.Wai.Handler.Warp.HTTP2.EncodeFrame where + +import Data.ByteString (ByteString) +import Network.HTTP2 + +---------------------------------------------------------------- + +goawayFrame :: StreamId -> ErrorCodeId -> ByteString -> ByteString +goawayFrame sid etype debugmsg = encodeFrame einfo frame + where + einfo = encodeInfo id 0 + frame = GoAwayFrame sid etype debugmsg + +resetFrame :: ErrorCodeId -> StreamId -> ByteString +resetFrame etype sid = encodeFrame einfo frame + where + einfo = encodeInfo id sid + frame = RSTStreamFrame etype + +settingsFrame :: (FrameFlags -> FrameFlags) -> SettingsList -> ByteString +settingsFrame func alist = encodeFrame einfo $ SettingsFrame alist + where + einfo = encodeInfo func 0 + +pingFrame :: ByteString -> ByteString +pingFrame bs = encodeFrame einfo $ PingFrame bs + where + einfo = encodeInfo setAck 0 + +windowUpdateFrame :: StreamId -> WindowSize -> ByteString +windowUpdateFrame sid winsiz = encodeFrame einfo $ WindowUpdateFrame winsiz + where + einfo = encodeInfo id sid diff --git a/Network/Wai/Handler/Warp/HTTP2/HPACK.hs b/Network/Wai/Handler/Warp/HTTP2/HPACK.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/HPACK.hs @@ -0,0 +1,49 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} + +module Network.Wai.Handler.Warp.HTTP2.HPACK where + +import Control.Arrow (first) +import qualified Control.Exception as E +import Data.ByteString.Builder (Builder) +import qualified Data.ByteString.Char8 as B8 +import Data.CaseInsensitive (foldedCase) +import Data.IORef (readIORef, writeIORef) +import Network.HPACK +import qualified Network.HTTP.Types as H +import Network.HTTP2 +import Network.Wai +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.Header +import Network.Wai.Handler.Warp.Response +import qualified Network.Wai.Handler.Warp.Settings as S +import Network.Wai.Handler.Warp.Types + +hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> Response + -> IO Builder +hpackEncodeHeader Context{encodeDynamicTable} ii settings rsp = do + hdr1 <- addServerAndDate hdr0 + let hdr2 = (":status", status) : map (first foldedCase) hdr1 + ehdrtbl <- readIORef encodeDynamicTable + (ehdrtbl', builder) <- encodeHeaderBuilder defaultEncodeStrategy ehdrtbl hdr2 + writeIORef encodeDynamicTable ehdrtbl' + return builder + where + hdr0 = responseHeaders rsp + status = B8.pack $ show $ H.statusCode $ responseStatus rsp + dc = dateCacher ii + rspidxhdr = indexResponseHeader hdr0 + defServer = S.settingsServerName settings + addServerAndDate = addDate dc rspidxhdr . addServer defServer rspidxhdr + + +---------------------------------------------------------------- + +hpackDecodeHeader :: HeaderBlockFragment -> Context -> IO HeaderList +hpackDecodeHeader hdrblk Context{decodeDynamicTable} = do + hdrtbl <- readIORef decodeDynamicTable + (hdrtbl', hdr) <- decodeHeader hdrtbl hdrblk `E.onException` cleanup + writeIORef decodeDynamicTable hdrtbl' + return hdr + where + cleanup = E.throwIO $ ConnectionError CompressionError "cannot decompress the header" diff --git a/Network/Wai/Handler/Warp/HTTP2/Manager.hs b/Network/Wai/Handler/Warp/HTTP2/Manager.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Manager.hs @@ -0,0 +1,86 @@ +{-# LANGUAGE CPP #-} + +-- | A thread pool manager. +-- The manager has responsibility to spawn and kill +-- worker threads. +module Network.Wai.Handler.Warp.HTTP2.Manager ( + Manager + , start + , setAction + , stop + , spawnAction + , replaceWithAction + ) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative +#endif +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad (void) +import Data.Set (Set) +import qualified Data.Set as Set +import Network.Wai.Handler.Warp.IORef + +---------------------------------------------------------------- + +data Command = Stop | Spawn | Replace ThreadId + +data Manager = Manager (TQueue Command) (IORef (IO ())) + +-- | Starting a thread pool manager. +-- Its action is initially set to 'return ()' and should be set +-- by 'setAction'. This allows that the action can include +-- the manager itself. +start :: IO Manager +start = do + tset <- newThreadSet + q <- newTQueueIO + ref <- newIORef (return ()) + void $ forkIO $ go q tset ref + return $ Manager q ref + where + go q tset ref = do + x <- atomically $ readTQueue q + case x of + Stop -> kill tset + Spawn -> next + Replace oldtid -> do + del tset oldtid + next + where + next = do + action <- readIORef ref + newtid <- forkIO $ action + add tset newtid + go q tset ref + +setAction :: Manager -> IO () -> IO () +setAction (Manager _ ref) action = writeIORef ref action + +stop :: Manager -> IO () +stop (Manager q _) = atomically $ writeTQueue q Stop + +spawnAction :: Manager -> IO () +spawnAction (Manager q _) = atomically $ writeTQueue q Spawn + +replaceWithAction :: Manager -> ThreadId -> IO () +replaceWithAction (Manager q _) tid = atomically $ writeTQueue q $ Replace tid + +---------------------------------------------------------------- + +newtype ThreadSet = ThreadSet (IORef (Set ThreadId)) + +newThreadSet :: IO ThreadSet +newThreadSet = ThreadSet <$> newIORef Set.empty + +add :: ThreadSet -> ThreadId -> IO () +add (ThreadSet ref) tid = + atomicModifyIORef' ref (\set -> (Set.insert tid set, ())) + +del :: ThreadSet -> ThreadId -> IO () +del (ThreadSet ref) tid = + atomicModifyIORef' ref (\set -> (Set.delete tid set, ())) + +kill :: ThreadSet -> IO () +kill (ThreadSet ref) = Set.toList <$> readIORef ref >>= mapM_ killThread diff --git a/Network/Wai/Handler/Warp/HTTP2/Receiver.hs b/Network/Wai/Handler/Warp/HTTP2/Receiver.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Receiver.hs @@ -0,0 +1,319 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE PatternGuards #-} + +module Network.Wai.Handler.Warp.HTTP2.Receiver (frameReceiver) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative +#endif +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad (when, unless, void) +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.Maybe (isJust) +import Network.HTTP2 +import Network.HTTP2.Priority +import Network.Wai.Handler.Warp.HTTP2.EncodeFrame +import Network.Wai.Handler.Warp.HTTP2.HPACK +import Network.Wai.Handler.Warp.HTTP2.Request +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.IORef +import Network.Wai.Handler.Warp.Types + +---------------------------------------------------------------- + +frameReceiver :: Context -> MkReq -> (BufSize -> IO ByteString) -> IO () +frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway + where + Context{ http2settings + , streamTable + , concurrency + , continued + , currentStreamId + , inputQ + , outputQ + } = ctx + sendGoaway e + | Just (ConnectionError err msg) <- E.fromException e = do + csid <- readIORef currentStreamId + let frame = goawayFrame csid err msg + enqueue outputQ (OGoaway frame) highestPriority + | otherwise = return () + + sendReset err sid = do + let frame = resetFrame err sid + enqueue outputQ (OFrame frame) highestPriority + + loop = do + hd <- recvN frameHeaderLength + if BS.null hd then + enqueue outputQ OFinish highestPriority + else do + cont <- processStreamGuardingError $ decodeFrameHeader hd + when cont loop + + processStreamGuardingError (_, FrameHeader{streamId}) + | isResponse streamId = E.throwIO $ ConnectionError ProtocolError "stream id should be odd" + processStreamGuardingError (FrameUnknown _, FrameHeader{payloadLength}) = do + mx <- readIORef continued + case mx of + Nothing -> do + -- ignoring unknown frame + consume payloadLength + return True + Just _ -> E.throwIO $ ConnectionError ProtocolError "unknown frame" + processStreamGuardingError (FramePushPromise, _) = + E.throwIO $ ConnectionError ProtocolError "push promise is not allowed" + processStreamGuardingError typhdr@(ftyp, header@FrameHeader{payloadLength}) = do + settings <- readIORef http2settings + case checkFrameHeader settings typhdr of + Left h2err -> case h2err of + StreamError err sid -> do + sendReset err sid + consume payloadLength + return True + connErr -> E.throwIO connErr + Right _ -> do + ex <- E.try $ controlOrStream ftyp header + case ex of + Left (StreamError err sid) -> do + sendReset err sid + return True + Left connErr -> E.throw connErr + Right cont -> return cont + + controlOrStream ftyp header@FrameHeader{streamId, payloadLength} + | isControl streamId = do + pl <- recvN payloadLength + control ftyp header pl ctx + | otherwise = do + checkContinued + strm@Stream{streamState,streamContentLength} <- getStream + pl <- recvN payloadLength + state <- readIORef streamState + state' <- stream ftyp header pl ctx state strm + case state' of + Open (NoBody hdr pri) -> do + resetContinued + case validateHeaders hdr of + Just vh -> do + when (isJust (vhCL vh) && vhCL vh /= Just 0) $ + E.throwIO $ StreamError ProtocolError streamId + writeIORef streamState HalfClosed + let req = mkreq vh (return "") + atomically $ writeTQueue inputQ $ Input strm req pri + Nothing -> E.throwIO $ StreamError ProtocolError streamId + Open (HasBody hdr pri) -> do + resetContinued + case validateHeaders hdr of + Just vh -> do + q <- newTQueueIO + writeIORef streamState (Open (Body q)) + writeIORef streamContentLength $ vhCL vh + readQ <- newReadBody q + bodySource <- mkSource readQ + let req = mkreq vh (readSource bodySource) + atomically $ writeTQueue inputQ $ Input strm req pri + Nothing -> E.throwIO $ StreamError ProtocolError streamId + s@(Open Continued{}) -> do + setContinued + writeIORef streamState s + s -> do -- Idle, Open Body, HalfClosed, Closed + resetContinued + writeIORef streamState s + return True + where + setContinued = writeIORef continued (Just streamId) + resetContinued = writeIORef continued Nothing + checkContinued = do + mx <- readIORef continued + case mx of + Nothing -> return () + Just sid + | sid == streamId && ftyp == FrameContinuation -> return () + | otherwise -> E.throwIO $ ConnectionError ProtocolError "continuation frame must follow" + getStream = do + mstrm0 <- search streamTable streamId + case mstrm0 of + Just strm0 -> do + when (ftyp == FrameHeaders) $ do + st <- readIORef $ streamState strm0 + when (isHalfClosed st) $ E.throwIO $ ConnectionError ProtocolError "header must not be sent to half closed" + return strm0 + Nothing -> do + when (ftyp `notElem` [FrameHeaders,FramePriority]) $ + E.throwIO $ ConnectionError ProtocolError "this frame is not allowed in an idel stream" + when (ftyp == FrameHeaders) $ do + csid <- readIORef currentStreamId + if streamId <= csid then + E.throwIO $ ConnectionError ProtocolError "stream identifier must not decrease" + else + writeIORef currentStreamId streamId + cnt <- readIORef concurrency + when (cnt >= recommendedConcurrency) $ + E.throwIO $ ConnectionError RefusedStream "over max concurrency" + ws <- initialWindowSize <$> readIORef http2settings + newstrm <- newStream streamId (fromIntegral ws) + when (ftyp == FrameHeaders) $ opened ctx newstrm + insert streamTable streamId newstrm + return newstrm + + consume = void . recvN + +---------------------------------------------------------------- + +control :: FrameTypeId -> FrameHeader -> ByteString -> Context -> IO Bool +control FrameSettings header@FrameHeader{flags} bs Context{http2settings, outputQ} = do + SettingsFrame alist <- guardIt $ decodeSettingsFrame header bs + case checkSettingsList alist of + Just x -> E.throwIO x + Nothing -> return () + unless (testAck flags) $ do + modifyIORef http2settings $ \old -> updateSettings old alist + let frame = settingsFrame setAck [] + enqueue outputQ (OFrame frame) highestPriority + return True + +control FramePing FrameHeader{flags} bs Context{outputQ} = + if testAck flags then + E.throwIO $ ConnectionError ProtocolError "the ack flag of this ping frame must not be set" + else do + let frame = pingFrame bs + enqueue outputQ (OFrame frame) defaultPriority + return True + +control FrameGoAway _ _ Context{outputQ} = do + enqueue outputQ OFinish highestPriority + return False + +control FrameWindowUpdate header bs Context{connectionWindow} = do + WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs + w <- (n +) <$> atomically (readTVar connectionWindow) + when (isWindowOverflow w) $ E.throwIO $ ConnectionError FlowControlError "control window should be less than 2^31" + atomically $ writeTVar connectionWindow w + return True + +control _ _ _ _ = + -- must not reach here + return False + +---------------------------------------------------------------- + +guardIt :: Either HTTP2Error a -> IO a +guardIt x = case x of + Left err -> E.throwIO err + Right frame -> return frame + +checkPriority :: Priority -> StreamId -> IO () +checkPriority p me + | dep == me = E.throwIO $ StreamError ProtocolError me + | otherwise = return () + where + dep = streamDependency p + +stream :: FrameTypeId -> FrameHeader -> ByteString -> Context -> StreamState -> Stream -> IO StreamState +stream FrameHeaders header@FrameHeader{flags} bs ctx (Open JustOpened) Stream{streamNumber} = do + HeadersFrame mp frag <- guardIt $ decodeHeadersFrame header bs + pri <- case mp of + Nothing -> return defaultPriority + Just p -> do + checkPriority p streamNumber + return p + let endOfStream = testEndStream flags + endOfHeader = testEndHeader flags + if endOfHeader then do + hdr <- hpackDecodeHeader frag ctx + return $ if endOfStream then + Open (NoBody hdr pri) + else + Open (HasBody hdr pri) + else do + let !siz = BS.length frag + return $ Open $ Continued [frag] siz 1 endOfStream pri + +stream FrameHeaders header@FrameHeader{flags} bs _ s@(Open (Body q)) _ = do + -- trailer is not supported. + -- let's read and ignore it. + HeadersFrame _ _ <- guardIt $ decodeHeadersFrame header bs + let endOfStream = testEndStream flags + if endOfStream then do + atomically $ writeTQueue q "" + return HalfClosed + else + return s + +stream FrameData + header@FrameHeader{flags,payloadLength,streamId} + bs + Context{outputQ} s@(Open (Body q)) + Stream{streamNumber,streamBodyLength,streamContentLength} = do + DataFrame body <- guardIt $ decodeDataFrame header bs + let endOfStream = testEndStream flags + len0 <- readIORef streamBodyLength + let !len = len0 + payloadLength + writeIORef streamBodyLength len + when (payloadLength /= 0) $ do + let frame1 = windowUpdateFrame 0 payloadLength + frame2 = windowUpdateFrame streamNumber payloadLength + frame = frame1 `BS.append` frame2 + enqueue outputQ (OFrame frame) highestPriority + atomically $ writeTQueue q body + if endOfStream then do + mcl <- readIORef streamContentLength + case mcl of + Nothing -> return () + Just cl -> when (cl /= len) $ E.throwIO $ StreamError ProtocolError streamId + atomically $ writeTQueue q "" + return HalfClosed + else + return s + +stream FrameContinuation FrameHeader{flags} frag ctx (Open (Continued rfrags siz n endOfStream pri)) _ = do + let endOfHeader = testEndHeader flags + rfrags' = frag : rfrags + siz' = siz + BS.length frag + n' = n + 1 + when (siz' > 51200) $ -- fixme: hard coding: 50K + E.throwIO $ ConnectionError EnhanceYourCalm "Header is too big" + when (n' > 10) $ -- fixme: hard coding + E.throwIO $ ConnectionError EnhanceYourCalm "Header is too fragmented" + if endOfHeader then do + let hdrblk = BS.concat $ reverse rfrags' + hdr <- hpackDecodeHeader hdrblk ctx + return $ if endOfStream then + Open (NoBody hdr pri) + else + Open (HasBody hdr pri) + else + return $ Open $ Continued rfrags' siz' n' endOfStream pri + +stream FrameContinuation _ _ _ _ _ = E.throwIO $ ConnectionError ProtocolError "continue frame cannot come here" + +stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s Stream{streamWindow} = do + WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs + w <- (n +) <$> atomically (readTVar streamWindow) + when (isWindowOverflow w) $ + E.throwIO $ StreamError FlowControlError streamId + atomically $ writeTVar streamWindow w + return s + +stream FrameRSTStream header bs ctx _ strm = do + RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs + let cc = Reset e + closed ctx strm cc + return $ Closed cc -- will be written to streamState again + +stream FramePriority header bs Context{outputQ} s Stream{streamNumber} = do + PriorityFrame p <- guardIt $ decodePriorityFrame header bs + checkPriority p streamNumber + prepare outputQ streamNumber p + return s + +-- this ordering is important +stream _ _ _ _ (Open Continued{}) _ = E.throwIO $ ConnectionError ProtocolError "an illegal frame follows header/continuation frames" +stream FrameData FrameHeader{streamId} _ _ _ _ = E.throwIO $ StreamError StreamClosed streamId +stream _ FrameHeader{streamId} _ _ _ _ = E.throwIO $ StreamError ProtocolError streamId diff --git a/Network/Wai/Handler/Warp/HTTP2/Request.hs b/Network/Wai/Handler/Warp/HTTP2/Request.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Request.hs @@ -0,0 +1,141 @@ +{-# LANGUAGE OverloadedStrings, CPP #-} + +module Network.Wai.Handler.Warp.HTTP2.Request ( + mkRequest + , newReadBody + , MkReq + , ValidHeaders(..) + , validateHeaders + ) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative ((<$>)) +#endif +import Control.Concurrent.STM +import Control.Monad (when) +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as B8 +import Data.CaseInsensitive (mk) +import Data.IORef (IORef, readIORef, newIORef, writeIORef) +import Data.Maybe (isJust) +#if __GLASGOW_HASKELL__ < 709 +import Data.Monoid (mempty) +#endif +import Data.Word8 (isUpper,_colon) +import Network.HPACK +import Network.HTTP.Types (RequestHeaders,hRange) +import qualified Network.HTTP.Types as H +import Network.Socket (SockAddr) +import Network.Wai +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.ReadInt +import qualified Network.Wai.Handler.Warp.Settings as S (Settings, settingsNoParsePath) +import Network.Wai.Internal (Request(..)) + +data ValidHeaders = ValidHeaders { + vhMethod :: ByteString + , vhPath :: ByteString + , vhAuth :: Maybe ByteString + , vhCL :: Maybe Int + , vhHeader :: RequestHeaders + } + +type MkReq = ValidHeaders -> IO ByteString -> Request + +mkRequest :: S.Settings -> SockAddr -> MkReq +mkRequest settings addr (ValidHeaders m p ma _ hdr) body = req + where + (unparsedPath,query) = B8.break (=='?') p + path = H.extractPath unparsedPath + req = Request { + requestMethod = m + , httpVersion = http2ver + , rawPathInfo = if S.settingsNoParsePath settings then unparsedPath else path + , pathInfo = H.decodePathSegments path + , rawQueryString = query + , queryString = H.parseQuery query + , requestHeaders = hdr + , isSecure = True + , remoteHost = addr + , requestBody = body + , vault = mempty + , requestBodyLength = ChunkedBody -- fixme + , requestHeaderHost = ma + , requestHeaderRange = lookup hRange hdr + } + +---------------------------------------------------------------- + +data Pseudo = Pseudo { + colonMethod :: !(Maybe ByteString) + , colonPath :: !(Maybe ByteString) + , colonAuth :: !(Maybe ByteString) + , contentLen :: !(Maybe ByteString) + } + +emptyPseudo :: Pseudo +emptyPseudo = Pseudo Nothing Nothing Nothing Nothing + +validateHeaders :: HeaderList -> Maybe ValidHeaders +validateHeaders hs = case pseudo hs (emptyPseudo,id) of + Just (Pseudo (Just m) (Just p) ma mcl, h) + -> Just $ ValidHeaders m p ma (readInt <$> mcl) h + _ -> Nothing + where + pseudo [] (p,b) = Just (p,b []) + pseudo h@((k,v):kvs) (p,b) + | k == ":method" = if isJust (colonMethod p) then + Nothing + else + pseudo kvs (p { colonMethod = Just v },b) + | k == ":path" = if isJust (colonPath p) then + Nothing + else + pseudo kvs (p { colonPath = Just v },b) + | k == ":authority" = if isJust (colonAuth p) then + Nothing + else + pseudo kvs (p { colonAuth = Just v },b) + | k == ":scheme" = pseudo kvs (p,b) -- fixme: how to store :scheme? + | isPseudo k = Nothing + | otherwise = normal h (p,b) + + normal [] (p,b) = Just (p,b []) + normal ((k,v):kvs) (p,b) + | isPseudo k = Nothing + | k == "connection" = Nothing + | k == "te" = if v == "trailers" then + normal kvs (p, b . ((mk k,v) :)) + else + Nothing + | k == "content-length" + = normal kvs (p { contentLen = Just v },b) + | k == "host" = if isJust (colonAuth p) then + normal kvs (p,b) + else + normal kvs (p { colonAuth = Just v },b) + | otherwise = case BS.find isUpper k of + Nothing -> normal kvs (p, b . ((mk k,v) :)) + Just _ -> Nothing + + isPseudo "" = False + isPseudo k = BS.head k == _colon + + +---------------------------------------------------------------- + +newReadBody :: TQueue ByteString -> IO (IO ByteString) +newReadBody q = do + ref <- newIORef False + return $ readBody q ref + +readBody :: TQueue ByteString -> IORef Bool -> IO ByteString +readBody q ref = do + eof <- readIORef ref + if eof then + return "" + else do + bs <- atomically $ readTQueue q + when (bs == "") $ writeIORef ref True + return bs diff --git a/Network/Wai/Handler/Warp/HTTP2/Sender.hs b/Network/Wai/Handler/Warp/HTTP2/Sender.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Sender.hs @@ -0,0 +1,398 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE BangPatterns, CPP #-} +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} + +module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative +#endif +import Control.Concurrent (forkIO) +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad (void, unless) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Builder.Extra as B +import Foreign.Ptr +import Network.HTTP2 +import Network.HTTP2.Priority +import Network.Wai +import Network.Wai.Handler.Warp.Buffer +import Network.Wai.Handler.Warp.HTTP2.EncodeFrame +import Network.Wai.Handler.Warp.HTTP2.HPACK +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.IORef +import qualified Network.Wai.Handler.Warp.Settings as S +import Network.Wai.Handler.Warp.Types +import Network.Wai.Internal (Response(..)) +import qualified System.PosixCompat.Files as P + +#ifdef WINDOWS +import qualified System.IO as IO +#else +import Network.Wai.Handler.Warp.FdCache +import Network.Wai.Handler.Warp.SendFile (positionRead) +import System.Posix.Types +#endif + +---------------------------------------------------------------- + +data Leftover = LZero + | LOne B.BufferWriter + | LTwo BS.ByteString B.BufferWriter + +---------------------------------------------------------------- + +unlessClosed :: Context -> Connection -> Stream -> IO () -> IO () +unlessClosed ctx + Connection{connSendAll} + strm@Stream{streamState,streamNumber} + body = E.handle resetStream $ do + state <- readIORef streamState + unless (isClosed state) body + where + resetStream e = do + closed ctx strm (ResetByMe e) + let rst = resetFrame InternalError streamNumber + connSendAll rst + +checkWindowSize :: TVar WindowSize -> TVar WindowSize -> PriorityTree Output -> Output -> Priority -> (WindowSize -> IO ()) -> IO () +checkWindowSize connWindow strmWindow outQ out pri body = do + cw <- atomically $ do + w <- readTVar connWindow + check (w > 0) + return w + sw <- atomically $ readTVar strmWindow + if sw == 0 then + void $ forkIO $ do + -- checkme: if connection is closed? GC throws dead lock error? + atomically $ do + x <- readTVar strmWindow + check (x > 0) + enqueue outQ out pri + else + body (min cw sw) + +frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO () +frameSender ctx@Context{outputQ,connectionWindow} + conn@Connection{connWriteBuffer,connBufferSize,connSendAll} + ii settings = go `E.catch` ignore + where + initialSettings = [(SettingsMaxConcurrentStreams,recommendedConcurrency)] + initialFrame = settingsFrame id initialSettings + bufHeaderPayload = connWriteBuffer `plusPtr` frameHeaderLength + headerPayloadLim = connBufferSize - frameHeaderLength + + go = do + connSendAll initialFrame + loop + + loop = dequeue outputQ >>= \(out, pri) -> switch out pri + + ignore :: E.SomeException -> IO () + ignore _ = return () + + switch OFinish _ = return () + switch (OGoaway frame) _ = connSendAll frame + switch (OFrame frame) _ = do + connSendAll frame + loop + switch out@(OResponse strm rsp aux) pri = unlessClosed ctx conn strm $ do + checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do + -- Header frame and Continuation frame + let sid = streamNumber strm + endOfStream = case aux of + Persist{} -> False + Oneshot hb -> not hb + len <- headerContinue sid rsp endOfStream + let total = len + frameHeaderLength + case aux of + Oneshot True -> do -- hasBody + -- Data frame payload + off <- sendHeadersIfNecessary total + Next datPayloadLen mnext <- fillResponseBodyGetNext conn ii off lim rsp + fillDataHeaderSend strm total datPayloadLen mnext pri + Oneshot False -> do + -- "closed" must be before "connSendAll". If not, + -- the context would be switched to the receiver, + -- resulting the inconsistency of concurrency. + closed ctx strm Finished + bufferIO connWriteBuffer total connSendAll + Persist sq tvar -> do + off <- sendHeadersIfNecessary total + Next datPayloadLen mnext <- fillStreamBodyGetNext conn off lim sq tvar + fillDataHeaderSend strm total datPayloadLen mnext pri + loop + switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do + checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do + -- Data frame payload + Next datPayloadLen mnext <- curr lim + fillDataHeaderSend strm 0 datPayloadLen mnext pri + loop + + headerContinue sid rsp endOfStream = do + builder <- hpackEncodeHeader ctx ii settings rsp + (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim + let flag0 = case signal of + B.Done -> setEndHeader defaultFlags + _ -> defaultFlags + flag = if endOfStream then setEndStream flag0 else flag0 + fillFrameHeader FrameHeaders len sid flag connWriteBuffer + continue sid len signal + + continue _ len B.Done = return len + continue sid len (B.More _ writer) = do + let total = len + frameHeaderLength + bufferIO connWriteBuffer total connSendAll + (len', signal') <- writer bufHeaderPayload headerPayloadLim + let flag = case signal' of + B.Done -> setEndHeader defaultFlags + _ -> defaultFlags + fillFrameHeader FrameContinuation len' sid flag connWriteBuffer + continue sid len' signal' + continue sid len (B.Chunk bs writer) = do + let total = len + frameHeaderLength + bufferIO connWriteBuffer total connSendAll + let (bs1,bs2) = BS.splitAt headerPayloadLim bs + len' = BS.length bs1 + void $ copy bufHeaderPayload bs1 + fillFrameHeader FrameContinuation len' sid defaultFlags connWriteBuffer + if bs2 == "" then + continue sid len' (B.More 0 writer) + else + continue sid len' (B.Chunk bs2 writer) + + sendHeadersIfNecessary total = do + let datPayloadOff = total + frameHeaderLength + if datPayloadOff < connBufferSize then + return datPayloadOff + else do + bufferIO connWriteBuffer total connSendAll + return frameHeaderLength + + fillDataHeaderSend strm otherLen datPayloadLen mnext pri = do + -- Data frame header + let sid = streamNumber strm + buf = connWriteBuffer `plusPtr` otherLen + total = otherLen + frameHeaderLength + datPayloadLen + flag = case mnext of + CFinish -> setEndStream defaultFlags + _ -> defaultFlags + fillFrameHeader FrameData datPayloadLen sid flag buf + -- "closed" must be before "connSendAll". If not, + -- the context would be switched to the receiver, + -- resulting the inconsistency of concurrency. + case mnext of + CFinish -> closed ctx strm Finished + _ -> return () + bufferIO connWriteBuffer total connSendAll + atomically $ do + modifyTVar' connectionWindow (subtract datPayloadLen) + modifyTVar' (streamWindow strm) (subtract datPayloadLen) + case mnext of + CFinish -> return () + CNext next -> enqueue outputQ (ONext strm next) pri + CNone -> return () + + fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf + where + hinfo = FrameHeader len flag sid + +---------------------------------------------------------------- + +{- +ResponseFile Status ResponseHeaders FilePath (Maybe FilePart) +ResponseBuilder Status ResponseHeaders Builder +ResponseStream Status ResponseHeaders StreamingBody +ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response +-} + +fillResponseBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Response -> IO Next +fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize} + _ off lim (ResponseBuilder _ _ bb) = do + let datBuf = connWriteBuffer `plusPtr` off + room = min (connBufferSize - off) lim + (len, signal) <- B.runBuilder bb datBuf room + return $ nextForBuilder connWriteBuffer connBufferSize len signal + +#ifdef WINDOWS +fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize} + _ off lim (ResponseFile _ _ path mpart) = do + let datBuf = connWriteBuffer `plusPtr` off + room = min (connBufferSize - off) lim + (start, bytes) <- fileStartEnd path mpart + -- fixme: how to close Handle? GC does it at this moment. + h <- IO.openBinaryFile path IO.ReadMode + IO.hSeek h IO.AbsoluteSeek start + len <- IO.hGetBufSome h datBuf (mini room bytes) + let bytes' = bytes - fromIntegral len + return $ nextForFile len connWriteBuffer connBufferSize h bytes' (return ()) +#else +fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize} + ii off lim (ResponseFile _ _ path mpart) = do + let Just fdcache = fdCacher ii + (fd, refresh) <- getFd fdcache path + let datBuf = connWriteBuffer `plusPtr` off + room = min (connBufferSize - off) lim + (start, bytes) <- fileStartEnd path mpart + len <- positionRead fd datBuf (mini room bytes) start + refresh + let len' = fromIntegral len + return $ nextForFile len connWriteBuffer connBufferSize fd (start + len') (bytes - len') refresh +#endif + +fillResponseBodyGetNext _ _ _ _ _ = error "fillResponseBodyGetNext" + +fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer) +fileStartEnd path Nothing = do + end <- fromIntegral . P.fileSize <$> P.getFileStatus path + return (0, end) +fileStartEnd _ (Just part) = + return (filePartOffset part, filePartByteCount part) + +---------------------------------------------------------------- + +fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> IO Next +fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize} + off lim sq tvar = do + let datBuf = connWriteBuffer `plusPtr` off + room = min (connBufferSize - off) lim + (leftover, cont, len) <- runStreamBuilder datBuf room sq + nextForStream connWriteBuffer connBufferSize sq tvar leftover cont len + +---------------------------------------------------------------- + +fillBufBuilder :: Buffer -> BufSize -> Leftover -> DynaNext +fillBufBuilder buf0 siz0 leftover lim = do + let payloadBuf = buf0 `plusPtr` frameHeaderLength + room = min (siz0 - frameHeaderLength) lim + case leftover of + LZero -> error "fillBufBuilder: LZero" + LOne writer -> do + (len, signal) <- writer payloadBuf room + getNext len signal + LTwo bs writer + | BS.length bs <= room -> do + buf1 <- copy payloadBuf bs + let len1 = BS.length bs + (len2, signal) <- writer buf1 (room - len1) + getNext (len1 + len2) signal + | otherwise -> do + let (bs1,bs2) = BS.splitAt room bs + void $ copy payloadBuf bs1 + getNext room (B.Chunk bs2 writer) + where + getNext l s = return $ nextForBuilder buf0 siz0 l s + +nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next +nextForBuilder _ _ len B.Done + = Next len CFinish +nextForBuilder buf siz len (B.More _ writer) + = Next len (CNext (fillBufBuilder buf siz (LOne writer))) +nextForBuilder buf siz len (B.Chunk bs writer) + = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer))) + +---------------------------------------------------------------- + +runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence + -> IO (Leftover, Bool, BytesFilled) +runStreamBuilder buf0 room0 sq = loop buf0 room0 0 + where + loop !buf !room !total = do + mbuilder <- atomically $ tryReadTBQueue sq + case mbuilder of + Nothing -> return (LZero, True, total) + Just (SBuilder builder) -> do + (len, signal) <- B.runBuilder builder buf room + let !total' = total + len + case signal of + B.Done -> loop (buf `plusPtr` len) (room - len) total' + B.More _ writer -> return (LOne writer, True, total') + B.Chunk bs writer -> return (LTwo bs writer, True, total') + Just SFlush -> return (LZero, True, total) + Just SFinish -> return (LZero, False, total) + +fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> DynaNext +fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do + let payloadBuf = buf0 `plusPtr` frameHeaderLength + room0 = min (siz0 - frameHeaderLength) lim0 + case leftover0 of + LZero -> do + (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq + getNext leftover cont len + LOne writer -> write writer payloadBuf room0 0 + LTwo bs writer + | BS.length bs <= room0 -> do + buf1 <- copy payloadBuf bs + let len = BS.length bs + write writer buf1 (room0 - len) len + | otherwise -> do + let (bs1,bs2) = BS.splitAt room0 bs + void $ copy payloadBuf bs1 + getNext (LTwo bs2 writer) True room0 + where + getNext = nextForStream buf0 siz0 sq tvar + write writer1 buf room sofar = do + (len, signal) <- writer1 buf room + case signal of + B.Done -> do + (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq + let !total = sofar + len + extra + getNext leftover cont total + B.More _ writer -> do + let !total = sofar + len + getNext (LOne writer) True total + B.Chunk bs writer -> do + let !total = sofar + len + getNext (LTwo bs writer) True total + +nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync + -> Leftover -> Bool -> BytesFilled + -> IO Next +nextForStream _ _ _ tvar _ False len = do + atomically $ writeTVar tvar SyncFinish + return $ Next len CFinish +nextForStream buf siz sq tvar LZero True len = do + atomically $ writeTVar tvar $ SyncNext (fillBufStream buf siz LZero sq tvar) + return $ Next len CNone +nextForStream buf siz sq tvar leftover True len = + return $ Next len (CNext (fillBufStream buf siz leftover sq tvar)) + +---------------------------------------------------------------- + +#ifdef WINDOWS +fillBufFile :: Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> DynaNext +fillBufFile buf siz h bytes refresh lim = do + let payloadBuf = buf `plusPtr` frameHeaderLength + room = min (siz - frameHeaderLength) lim + len <- IO.hGetBufSome h payloadBuf room + refresh + let bytes' = bytes - fromIntegral len + return $ nextForFile len buf siz h bytes' refresh + +nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next +nextForFile 0 _ _ _ _ _ = Next 0 CFinish +nextForFile len _ _ _ 0 _ = Next len CFinish +nextForFile len buf siz h bytes refresh = + Next len (CNext (fillBufFile buf siz h bytes refresh)) +#else +fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext +fillBufFile buf siz fd start bytes refresh lim = do + let payloadBuf = buf `plusPtr` frameHeaderLength + room = min (siz - frameHeaderLength) lim + len <- positionRead fd payloadBuf (mini room bytes) start + let len' = fromIntegral len + refresh + return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh + +nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next +nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish +nextForFile len _ _ _ _ 0 _ = Next len CFinish +nextForFile len buf siz fd start bytes refresh = + Next len (CNext (fillBufFile buf siz fd start bytes refresh)) +#endif + +mini :: Int -> Integer -> Int +mini i n + | fromIntegral i < n = i + | otherwise = fromIntegral n diff --git a/Network/Wai/Handler/Warp/HTTP2/Types.hs b/Network/Wai/Handler/Warp/HTTP2/Types.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Types.hs @@ -0,0 +1,229 @@ +{-# LANGUAGE OverloadedStrings, CPP #-} +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} + +module Network.Wai.Handler.Warp.HTTP2.Types where + +import Blaze.ByteString.Builder (Builder) +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative ((<$>),(<*>)) +#endif +import Control.Concurrent.STM +import Control.Exception (SomeException) +import Control.Monad (void) +import Control.Reaper +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.IntMap.Strict (IntMap, IntMap) +import qualified Data.IntMap.Strict as M +import qualified Network.HTTP.Types as H +import Network.Wai (Request, Response) +import Network.Wai.Handler.Warp.IORef +import Network.Wai.Handler.Warp.Types + +import Network.HTTP2 +import Network.HTTP2.Priority +import Network.HPACK + +---------------------------------------------------------------- + +http2ver :: H.HttpVersion +http2ver = H.HttpVersion 2 0 + +isHTTP2 :: Transport -> Bool +isHTTP2 TCP = False +isHTTP2 tls = useHTTP2 + where + useHTTP2 = case tlsNegotiatedProtocol tls of + Nothing -> False + Just proto -> "h2-" `BS.isPrefixOf` proto + +---------------------------------------------------------------- + +data Input = Input Stream Request Priority + +---------------------------------------------------------------- + +data Control a = CFinish + | CNext a + | CNone + +instance Show (Control a) where + show CFinish = "CFinish" + show (CNext _) = "CNext" + show CNone = "CNone" + +type DynaNext = WindowSize -> IO Next + +type BytesFilled = Int + +data Next = Next BytesFilled (Control DynaNext) + +data Output = OFinish + | OGoaway ByteString + | OFrame ByteString + | OResponse Stream Response Aux + | ONext Stream DynaNext + +---------------------------------------------------------------- + +data Sequence = SFinish + | SFlush + | SBuilder Builder + +data Sync = SyncNone + | SyncFinish + | SyncNext DynaNext + +data Aux = Oneshot Bool + | Persist (TBQueue Sequence) (TVar Sync) + +---------------------------------------------------------------- + +-- | The context for HTTP/2 connection. +data Context = Context { + http2settings :: IORef Settings + , streamTable :: StreamTable + , concurrency :: IORef Int + -- | RFC 7540 says "Other frames (from any stream) MUST NOT + -- occur between the HEADERS frame and any CONTINUATION + -- frames that might follow". This field is used to implement + -- this requirement. + , continued :: IORef (Maybe StreamId) + , currentStreamId :: IORef StreamId + , inputQ :: TQueue Input + , outputQ :: PriorityTree Output + , encodeDynamicTable :: IORef DynamicTable + , decodeDynamicTable :: IORef DynamicTable + , connectionWindow :: TVar WindowSize + } + +---------------------------------------------------------------- + +newContext :: IO Context +newContext = Context <$> newIORef defaultSettings + <*> initialize 10 -- fixme: hard coding: 10 + <*> newIORef 0 + <*> newIORef Nothing + <*> newIORef 0 + <*> newTQueueIO + <*> newPriorityTree + <*> (newDynamicTableForEncoding defaultDynamicTableSize >>= newIORef) + <*> (newDynamicTableForDecoding defaultDynamicTableSize >>= newIORef) + <*> newTVarIO defaultInitialWindowSize + +clearContext :: Context -> IO () +clearContext ctx = void $ reaperStop $ streamTable ctx + +---------------------------------------------------------------- + +data OpenState = + JustOpened + | Continued [HeaderBlockFragment] + Int -- Total size + Int -- The number of continuation frames + Bool -- End of stream + Priority + | NoBody HeaderList Priority + | HasBody HeaderList Priority + | Body (TQueue ByteString) + +data ClosedCode = Finished + | Killed + | Reset ErrorCodeId + | ResetByMe SomeException + deriving Show + +data StreamState = + Idle + | Open OpenState + | HalfClosed + | Closed ClosedCode + +isIdle :: StreamState -> Bool +isIdle Idle = True +isIdle _ = False + +isOpen :: StreamState -> Bool +isOpen Open{} = True +isOpen _ = False + +isHalfClosed :: StreamState -> Bool +isHalfClosed HalfClosed = True +isHalfClosed _ = False + +isClosed :: StreamState -> Bool +isClosed Closed{} = True +isClosed _ = False + +instance Show StreamState where + show Idle = "Idle" + show Open{} = "Open" + show HalfClosed = "HalfClosed" + show (Closed e) = "Closed: " ++ show e + +---------------------------------------------------------------- + +data Stream = Stream { + streamNumber :: StreamId + , streamState :: IORef StreamState + -- Next two fields are for error checking. + , streamContentLength :: IORef (Maybe Int) + , streamBodyLength :: IORef Int + , streamWindow :: TVar WindowSize + } + +instance Show Stream where + show s = show (streamNumber s) + +newStream :: StreamId -> WindowSize -> IO Stream +newStream sid win = Stream sid <$> newIORef Idle + <*> newIORef Nothing + <*> newIORef 0 + <*> newTVarIO win + +---------------------------------------------------------------- + +opened :: Context -> Stream -> IO () +opened Context{concurrency} Stream{streamState} = do + atomicModifyIORef' concurrency (\x -> ((x+1),())) + writeIORef streamState (Open JustOpened) + +closed :: Context -> Stream -> ClosedCode -> IO () +closed Context{concurrency} Stream{streamState} cc = do + atomicModifyIORef' concurrency (\x -> ((x-1),())) + writeIORef streamState (Closed cc) + +---------------------------------------------------------------- + +type StreamTable = Reaper (IntMap Stream) (M.Key, Stream) + +initialize :: Int -> IO StreamTable +initialize duration = mkReaper settings + where + settings = defaultReaperSettings { + reaperAction = clean + , reaperDelay = duration * 1000000 + , reaperCons = uncurry M.insert + , reaperNull = M.null + , reaperEmpty = M.empty + } + +clean :: IntMap Stream -> IO (IntMap Stream -> IntMap Stream) +clean old = do + new <- M.fromAscList <$> prune oldlist [] + return $ M.union new + where + oldlist = M.toDescList old + prune [] lst = return lst + prune (x@(_,s):xs) lst = do + st <- readIORef (streamState s) + if isClosed st then + prune xs lst + else + prune xs (x:lst) + +insert :: StreamTable -> M.Key -> Stream -> IO () +insert strmtbl k v = reaperAdd strmtbl (k,v) + +search :: StreamTable -> M.Key -> IO (Maybe Stream) +search strmtbl k = M.lookup k <$> reaperRead strmtbl diff --git a/Network/Wai/Handler/Warp/HTTP2/Worker.hs b/Network/Wai/Handler/Warp/HTTP2/Worker.hs new file mode 100644 --- /dev/null +++ b/Network/Wai/Handler/Warp/HTTP2/Worker.hs @@ -0,0 +1,185 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} +{-# LANGUAGE PatternGuards, BangPatterns #-} +{-# LANGUAGE CPP #-} + +module Network.Wai.Handler.Warp.HTTP2.Worker ( + Responder + , response + , worker + ) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative +#endif +import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception (Exception, SomeException(..), AsyncException(..)) +import qualified Control.Exception as E +import Control.Monad (void, when) +import Data.Typeable +import qualified Network.HTTP.Types as H +import Network.HTTP2 +import Network.HTTP2.Priority +import Network.Wai +import Network.Wai.Handler.Warp.HTTP2.EncodeFrame +import Network.Wai.Handler.Warp.HTTP2.Manager +import Network.Wai.Handler.Warp.HTTP2.Types +import Network.Wai.Handler.Warp.IORef +import qualified Network.Wai.Handler.Warp.Response as R +import qualified Network.Wai.Handler.Warp.Settings as S +import qualified Network.Wai.Handler.Warp.Timeout as T +import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceived(..)) + +---------------------------------------------------------------- + +-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'. +-- This type implements the second argument (Response -> IO ResponseReceived) +-- with extra arguments. +type Responder = ThreadContinue -> T.Handle -> Stream -> Priority -> Request -> + Response -> IO ResponseReceived + +-- | This function is passed to workers. +-- They also pass 'Response's from 'Application's to this function. +-- This function enqueues commands for the HTTP/2 sender. +response :: Context -> Manager -> Responder +response Context{outputQ} mgr tconf th strm pri req rsp = do + case rsp of + ResponseStream _ _ strmbdy -> do + -- We must not exit this WAI application. + -- If the application exits, streaming would be also closed. + -- So, this work occupies this thread. + -- + -- We need to increase the number of workers. + myThreadId >>= replaceWithAction mgr + -- After this work, this thread stops to decease + -- the number of workers. + setThreadContinue tconf False + -- Since 'StreamingBody' is loop, we cannot control it. + -- So, let's serialize 'Builder' with a designated queue. + sq <- newTBQueueIO 10 -- fixme: hard coding: 10 + tvar <- newTVarIO SyncNone + enqueue outputQ (OResponse strm rsp (Persist sq tvar)) pri + let push b = do + atomically $ writeTBQueue sq (SBuilder b) + T.tickle th + flush = atomically $ writeTBQueue sq SFlush + -- Since we must not enqueue an empty queue to the priority + -- queue, we spawn a thread to ensure that the designated + -- queue is not empty. + void $ forkIO $ waiter tvar sq (enqueue outputQ) strm pri + strmbdy push flush + atomically $ writeTBQueue sq SFinish + _ -> do + setThreadContinue tconf True + let hasBody = requestMethod req /= H.methodHead + || R.hasBody (responseStatus rsp) + enqueue outputQ (OResponse strm rsp (Oneshot hasBody)) pri + return ResponseReceived + +data Break = Break deriving (Show, Typeable) + +instance Exception Break + +worker :: Context -> S.Settings -> T.Manager -> Application -> Responder -> IO () +worker ctx@Context{inputQ,outputQ} set tm app responder = do + tid <- myThreadId + sinfo <- newStreamInfo + tcont <- newThreadContinue + let setup = T.register tm $ E.throwTo tid Break + E.bracket setup T.cancel $ go sinfo tcont + where + go sinfo tcont th = do + setThreadContinue tcont True + ex <- E.try $ do + T.pause th + Input strm req pri <- atomically $ readTQueue inputQ + setStreamInfo sinfo strm req + T.resume th + T.tickle th + app req $ responder tcont th strm pri req + cont1 <- case ex of + Right ResponseReceived -> return True + Left e@(SomeException _) + | Just Break <- E.fromException e -> do + cleanup sinfo Nothing + return True + -- killed by the sender + | Just ThreadKilled <- E.fromException e -> do + cleanup sinfo Nothing + return False + | otherwise -> do + cleanup sinfo (Just e) + return True + cont2 <- getThreadContinue tcont + when (cont1 && cont2) $ go sinfo tcont th + cleanup sinfo me = do + m <- getStreamInfo sinfo + case m of + Nothing -> return () + Just (strm,req) -> do + closed ctx strm Killed + let frame = resetFrame InternalError (streamNumber strm) + enqueue outputQ (OFrame frame) highestPriority + case me of + Nothing -> return () + Just e -> S.settingsOnException set (Just req) e + clearStreamInfo sinfo + +waiter :: TVar Sync -> TBQueue Sequence + -> (Output -> Priority -> IO ()) -> Stream -> Priority + -> IO () +waiter tvar sq enq strm pri = do + mx <- atomically $ do + mout <- readTVar tvar + case mout of + SyncNone -> retry + SyncNext nxt -> do + writeTVar tvar SyncNone + return $ Just nxt + SyncFinish -> return Nothing + case mx of + Nothing -> return () + Just next -> do + atomically $ do + isEmpty <- isEmptyTBQueue sq + when isEmpty retry + enq (ONext strm next) pri + waiter tvar sq enq strm pri + +---------------------------------------------------------------- + +-- | It would nice if responders could return values to workers. +-- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0. +-- It is not wise to change this type. +-- So, a reference is shared by a responder and its worker. +-- The reference refers a value of this type as a return value. +-- If 'True', the worker continue to serve requests. +-- Otherwise, the worker get finished. +newtype ThreadContinue = ThreadContinue (IORef Bool) + +newThreadContinue :: IO ThreadContinue +newThreadContinue = ThreadContinue <$> newIORef True + +setThreadContinue :: ThreadContinue -> Bool -> IO () +setThreadContinue (ThreadContinue ref) x = writeIORef ref x + +getThreadContinue :: ThreadContinue -> IO Bool +getThreadContinue (ThreadContinue ref) = readIORef ref + +---------------------------------------------------------------- + +-- | The type to store enough information for 'settingsOnException'. +newtype StreamInfo = StreamInfo (IORef (Maybe (Stream,Request))) + +newStreamInfo :: IO StreamInfo +newStreamInfo = StreamInfo <$> newIORef Nothing + +clearStreamInfo :: StreamInfo -> IO () +clearStreamInfo (StreamInfo ref) = writeIORef ref Nothing + +setStreamInfo :: StreamInfo -> Stream -> Request -> IO () +setStreamInfo (StreamInfo ref) strm req = writeIORef ref $ Just (strm,req) + +getStreamInfo :: StreamInfo -> IO (Maybe (Stream, Request)) +getStreamInfo (StreamInfo ref) = readIORef ref diff --git a/Network/Wai/Handler/Warp/IO.hs b/Network/Wai/Handler/Warp/IO.hs --- a/Network/Wai/Handler/Warp/IO.hs +++ b/Network/Wai/Handler/Warp/IO.hs @@ -3,9 +3,9 @@ module Network.Wai.Handler.Warp.IO where -import Data.ByteString.Internal (ByteString(..)) -import Foreign.ForeignPtr (newForeignPtr_) +import Data.ByteString (ByteString) import Network.Wai.Handler.Warp.Buffer +import Network.Wai.Handler.Warp.Types -- Support for doctest, where cabal macros are not available #ifndef MIN_VERSION_blaze_builder @@ -21,7 +21,7 @@ toBufIOWith buf !size io builder = loop firstWriter where firstWriter = runBuilder builder - runIO len = toBS buf len >>= io + runIO len = bufferIO buf len io loop writer = do (len, signal) <- writer buf size case signal of @@ -63,8 +63,3 @@ loop next #endif /* !MIN_VERSION_blaze_builder(0,4,0) */ - -toBS :: Buffer -> Int -> IO ByteString -toBS ptr siz = do - fptr <- newForeignPtr_ ptr - return $ PS fptr 0 siz diff --git a/Network/Wai/Handler/Warp/Internal.hs b/Network/Wai/Handler/Warp/Internal.hs --- a/Network/Wai/Handler/Warp/Internal.hs +++ b/Network/Wai/Handler/Warp/Internal.hs @@ -1,21 +1,78 @@ {-# OPTIONS_GHC -fno-warn-deprecations #-} -module Network.Wai.Handler.Warp.Internal - ( Settings (..) - , getOnOpen - , getOnClose - , getOnException - ) where -import Network.Wai.Handler.Warp.Settings (Settings (..)) -import Network.Socket (SockAddr) -import Network.Wai (Request) -import Control.Exception (SomeException) - -getOnOpen :: Settings -> SockAddr -> IO Bool -getOnOpen = settingsOnOpen - -getOnClose :: Settings -> SockAddr -> IO () -getOnClose = settingsOnClose +module Network.Wai.Handler.Warp.Internal ( + -- * Settings + Settings (..) + , ProxyProtocol(..) + -- * Low level run functions + , runSettingsConnection + , runSettingsConnectionMaker + , runSettingsConnectionMakerSecure + , Transport (..) + -- * Connection + , Connection (..) + , socketConnection + -- ** Receive + , Recv + , RecvBuf + , makePlainReceiveN + -- ** Buffer + , Buffer + , BufSize + , bufferSize + , allocateBuffer + , freeBuffer + , copy + -- ** Sendfile + , FileId (..) + , SendFile + , sendFile + , readSendFile + -- * Version + , warpVersion + -- * Data types + , InternalInfo (..) + , HeaderValue + , IndexedHeader + , requestMaxIndex + -- * Time out manager + -- | + -- + -- In order to provide slowloris protection, Warp provides timeout handlers. We + -- follow these rules: + -- + -- * A timeout is created when a connection is opened. + -- + -- * When all request headers are read, the timeout is tickled. + -- + -- * Every time at least 2048 bytes of the request body are read, the timeout + -- is tickled. + -- + -- * The timeout is paused while executing user code. This will apply to both + -- the application itself, and a ResponseSource response. The timeout is + -- resumed as soon as we return from user code. + -- + -- * Every time data is successfully sent to the client, the timeout is tickled. + , module Network.Wai.Handler.Warp.Timeout + -- * File descriptor cache + , module Network.Wai.Handler.Warp.FdCache + -- * Date + , module Network.Wai.Handler.Warp.Date + -- * Request and response + , Source + , recvRequest + , sendResponse + ) where -getOnException :: Settings -> Maybe Request -> SomeException -> IO () -getOnException = settingsOnException +import Network.Wai.Handler.Warp.Buffer +import Network.Wai.Handler.Warp.Date +import Network.Wai.Handler.Warp.FdCache +import Network.Wai.Handler.Warp.Header +import Network.Wai.Handler.Warp.Recv +import Network.Wai.Handler.Warp.Request +import Network.Wai.Handler.Warp.Response +import Network.Wai.Handler.Warp.Run +import Network.Wai.Handler.Warp.SendFile +import Network.Wai.Handler.Warp.Settings +import Network.Wai.Handler.Warp.Timeout +import Network.Wai.Handler.Warp.Types diff --git a/Network/Wai/Handler/Warp/MultiMap.hs b/Network/Wai/Handler/Warp/MultiMap.hs --- a/Network/Wai/Handler/Warp/MultiMap.hs +++ b/Network/Wai/Handler/Warp/MultiMap.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE CPP #-} + module Network.Wai.Handler.Warp.MultiMap ( MMap , Some(..) @@ -16,7 +18,9 @@ , merge ) where +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative ((<$>)) +#endif import Data.List (foldl') ---------------------------------------------------------------- diff --git a/Network/Wai/Handler/Warp/Recv.hs b/Network/Wai/Handler/Warp/Recv.hs --- a/Network/Wai/Handler/Warp/Recv.hs +++ b/Network/Wai/Handler/Warp/Recv.hs @@ -1,23 +1,31 @@ -{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings #-} {-# LANGUAGE CPP #-} module Network.Wai.Handler.Warp.Recv ( receive + , receiveBuf + , makeReceiveN + , makePlainReceiveN + , spell ) where +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative ((<$>)) -import Control.Monad (void) -import qualified Data.ByteString as BS (empty) -import Data.ByteString.Internal (ByteString(..), mallocByteString) +#endif +import qualified Control.Exception as E +import qualified Data.ByteString as BS +import Data.ByteString.Internal (ByteString(..)) import Data.Word (Word8) import Foreign.C.Error (eAGAIN, getErrno, throwErrno) import Foreign.C.Types import Foreign.ForeignPtr (withForeignPtr) -import Foreign.Ptr (Ptr, castPtr) +import Foreign.Ptr (Ptr, castPtr, plusPtr) import GHC.Conc (threadWaitRead) import Network.Socket (Socket, fdSocket) -import System.Posix.Types (Fd(..)) import Network.Wai.Handler.Warp.Buffer +import Network.Wai.Handler.Warp.IORef +import Network.Wai.Handler.Warp.Types +import System.Posix.Types (Fd(..)) #ifdef mingw32_HOST_OS import GHC.IO.FD (FD(..), readRawBufferPtr) @@ -26,37 +34,95 @@ ---------------------------------------------------------------- -receive :: Socket -> Buffer -> Int -> IO ByteString -receive sock buf size = do - bytes <- fromIntegral <$> receiveloop sock' buf' size' - if bytes == 0 then - return BS.empty - else do - fptr <- mallocByteString bytes - void $ withForeignPtr fptr $ \ptr -> - c_memcpy ptr buf (fromIntegral bytes) - return $! PS fptr 0 bytes +makeReceiveN :: ByteString -> Recv -> RecvBuf -> IO (BufSize -> IO ByteString) +makeReceiveN bs0 recv recvBuf = do + ref <- newIORef bs0 + return $ receiveN ref recv recvBuf + +-- | This function returns a receiving function +-- based on two receiving functions. +-- The returned function efficiently manages received data +-- which is initialized by the first argument. +-- The returned function may allocate a byte string with malloc(). +makePlainReceiveN :: Socket -> ByteString -> IO (BufSize -> IO ByteString) +makePlainReceiveN s bs0 = do + ref <- newIORef bs0 + pool <- newBufferPool + return $ receiveN ref (receive s pool) (receiveBuf s) + +receiveN :: IORef ByteString -> Recv -> RecvBuf -> BufSize -> IO ByteString +receiveN ref recv recvBuf size = E.handle handler $ do + cached <- readIORef ref + (bs, leftover) <- spell cached size recv recvBuf + writeIORef ref leftover + return bs + where + handler :: E.SomeException -> IO ByteString + handler _ = return "" + +---------------------------------------------------------------- + +spell :: ByteString -> BufSize -> IO ByteString -> RecvBuf -> IO (ByteString, ByteString) +spell init0 siz0 recv recvBuf + | siz0 <= len0 = return $ BS.splitAt siz0 init0 + -- fixme: hard coding 4096 + | siz0 <= 4096 = loop [init0] (siz0 - len0) + | otherwise = do + bs@(PS fptr _ _) <- mallocBS siz0 + withForeignPtr fptr $ \ptr -> do + ptr' <- copy ptr init0 + full <- recvBuf ptr' (siz0 - len0) + if full then + return (bs, "") + else + return ("", "") -- fixme where - sock' = fdSocket sock - buf' = castPtr buf - size' = fromIntegral size + len0 = BS.length init0 + loop bss siz = do + bs <- recv + let len = BS.length bs + if len == 0 then + return ("", "") + else if len >= siz then do + let (consume, leftover) = BS.splitAt siz bs + ret = BS.concat $ reverse (consume : bss) + return (ret, leftover) + else do + let bss' = bs : bss + siz' = siz - len + loop bss' siz' -#ifdef mingw32_HOST_OS +receive :: Socket -> BufferPool -> Recv +receive sock pool = withBufferPool pool $ \ (ptr, size) -> do + let sock' = fdSocket sock + size' = fromIntegral size + fromIntegral <$> receiveloop sock' ptr size' + +receiveBuf :: Socket -> RecvBuf +receiveBuf sock buf0 siz0 = loop buf0 siz0 + where + loop _ 0 = return True + loop buf siz = do + n <- fromIntegral <$> receiveloop fd buf (fromIntegral siz) + -- fixme: what should we do in the case of n == 0 + if n == 0 then + return False + else + loop (buf `plusPtr` n) (siz - n) + fd = fdSocket sock + receiveloop :: CInt -> Ptr Word8 -> CSize -> IO CInt -#else -receiveloop :: CInt -> Ptr CChar -> CSize -> IO CInt -#endif -receiveloop sock buf size = do +receiveloop sock ptr size = do #ifdef mingw32_HOST_OS - bytes <- windowsThreadBlockHack $ fmap fromIntegral $ readRawBufferPtr "recv" (FD sock 1) buf 0 size + bytes <- windowsThreadBlockHack $ fmap fromIntegral $ readRawBufferPtr "recv" (FD sock 1) (castPtr ptr) 0 size #else - bytes <- c_recv sock buf size 0 + bytes <- c_recv sock (castPtr ptr) size 0 #endif if bytes == -1 then do errno <- getErrno if errno == eAGAIN then do threadWaitRead (Fd sock) - receiveloop sock buf size + receiveloop sock ptr size else throwErrno "receiveloop" else @@ -65,6 +131,3 @@ -- fixme: the type of the return value foreign import ccall unsafe "recv" c_recv :: CInt -> Ptr CChar -> CSize -> CInt -> IO CInt - -foreign import ccall unsafe "string.h memcpy" - c_memcpy :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8) diff --git a/Network/Wai/Handler/Warp/Request.hs b/Network/Wai/Handler/Warp/Request.hs --- a/Network/Wai/Handler/Warp/Request.hs +++ b/Network/Wai/Handler/Warp/Request.hs @@ -98,7 +98,7 @@ headerLines src = do bs <- readSource src if S.null bs - then throwIO $ NotEnoughLines [] + then throwIO ConnectionClosedByPeer else push src (THStatus 0 id id) bs ---------------------------------------------------------------- diff --git a/Network/Wai/Handler/Warp/RequestHeader.hs b/Network/Wai/Handler/Warp/RequestHeader.hs --- a/Network/Wai/Handler/Warp/RequestHeader.hs +++ b/Network/Wai/Handler/Warp/RequestHeader.hs @@ -51,6 +51,8 @@ -- *** Exception: Warp: Invalid first line of request: "GET " -- >>> parseRequestLine "GET /NotHTTP UNKNOWN/1.1" -- *** Exception: Warp: Request line specified a non-HTTP request +-- >>> parseRequestLine "PRI * HTTP/2.0" +-- ("PRI","*","",HTTP/2.0) parseRequestLine :: ByteString -> IO (H.Method ,ByteString -- Path @@ -101,12 +103,13 @@ check httpptr 4 47 -- '/' check httpptr 6 46 -- '.' httpVersion httpptr = do - major <- peek $ httpptr `plusPtr` 5 - minor <- peek $ httpptr `plusPtr` 7 - return $ if major == (49 :: Word8) && minor == (49 :: Word8) then - H.http11 - else - H.http10 + major <- peek (httpptr `plusPtr` 5) :: IO Word8 + minor <- peek (httpptr `plusPtr` 7) :: IO Word8 + let version + | major == 49 = if minor == 49 then H.http11 else H.http10 + | major == 50 && minor == 48 = H.HttpVersion 2 0 + | otherwise = H.http10 + return version bs ptr p0 p1 = PS fptr o l where o = p0 `minusPtr` ptr @@ -127,7 +130,7 @@ parseHeader :: ByteString -> H.Header parseHeader s = - let (k, rest) = S.breakByte 58 s -- ':' + let (k, rest) = S.break (== 58) s -- ':' rest' = S.dropWhile (\c -> c == 32 || c == 9) $ S.drop 1 rest in (CI.mk k, rest') @@ -146,7 +149,7 @@ case B.readInteger bs4 of Just (j, bs5) | j >= i -> Just (HH.ByteRangeFromTo i j, bs5) _ -> Just (HH.ByteRangeFrom i, bs4) - ranges front bs3 + ranges front bs3 | S.null bs3 = Just (front []) | otherwise = do bs4 <- stripPrefix "," bs3 diff --git a/Network/Wai/Handler/Warp/Response.hs b/Network/Wai/Handler/Warp/Response.hs --- a/Network/Wai/Handler/Warp/Response.hs +++ b/Network/Wai/Handler/Warp/Response.hs @@ -9,6 +9,9 @@ , fileRange -- for testing , warpVersion , defaultServerValue + , addDate + , addServer + , hasBody ) where #ifndef MIN_VERSION_base @@ -17,7 +20,9 @@ import Blaze.ByteString.Builder (fromByteString, Builder, flush) import Blaze.ByteString.Builder.HTTP (chunkedTransferEncoding, chunkedTransferTerminator) +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative +#endif import Control.Exception import Data.Array ((!)) import Data.ByteString (ByteString) @@ -30,19 +35,23 @@ import Data.List (deleteBy) import Data.Maybe (isJust, listToMaybe) #if MIN_VERSION_base(4,5,0) -import Data.Monoid ((<>), mempty) +# if __GLASGOW_HASKELL__ < 709 +import Data.Monoid (mempty) +# endif +import Data.Monoid ((<>)) #else import Data.Monoid (mappend, mempty) #endif import Data.Version (showVersion) import qualified Network.HTTP.Types as H import Network.Wai -import qualified Network.Wai.Handler.Warp.Date as D import Network.Wai.Handler.Warp.Buffer (toBlazeBuffer) +import qualified Network.Wai.Handler.Warp.Date as D +import qualified Network.Wai.Handler.Warp.FdCache as F import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.IO (toBufIOWith) -import Network.Wai.Handler.Warp.ResponseHeader import Network.Wai.Handler.Warp.RequestHeader (parseByteRanges) +import Network.Wai.Handler.Warp.ResponseHeader import qualified Network.Wai.Handler.Warp.Timeout as T import Network.Wai.Handler.Warp.Types import Network.Wai.Internal @@ -174,9 +183,9 @@ -> IO Bool -- ^ Returing True if the connection is persistent. sendResponse defServer conn ii req reqidxhdr src response = do hs <- addServerAndDate hs0 - if hasBody s req then do + if hasBody s then do -- HEAD comes here even if it does not have body. - sendRsp conn ver s hs rsp + sendRsp conn mfdc ver s hs rsp T.tickle th return ret else do @@ -190,10 +199,11 @@ rspidxhdr = indexResponseHeader hs0 th = threadHandle ii dc = dateCacher ii + mfdc = fdCacher ii addServerAndDate = addDate dc rspidxhdr . addServer defServer rspidxhdr mRange = reqidxhdr ! idxRange (isPersist,isChunked0) = infoFromRequest req reqidxhdr - isChunked = if isHead then False else isChunked0 + isChunked = not isHead && isChunked0 (isKeepAlive, needsChunked) = infoFromResponse rspidxhdr (isPersist,isChunked) isHead = requestMethod req == H.methodHead rsp = case response of @@ -217,28 +227,41 @@ ---------------------------------------------------------------- sendRsp :: Connection + -> Maybe F.MutableFdCache -> H.HttpVersion -> H.Status -> H.ResponseHeaders -> Rsp -> IO () -sendRsp conn ver s0 hs0 (RspFile path mPart mRange isHead hook) = do +sendRsp conn mfdc ver s0 hs0 (RspFile path mPart mRange isHead hook) = do ex <- fileRange s0 hs path mPart mRange case ex of Left _ex -> #ifdef WARP_DEBUG print _ex >> #endif - sendRsp conn ver s2 hs2 (RspBuilder body True) + sendRsp conn mfdc ver s2 hs2 (RspBuilder body True) Right (s, hs1, beg, len) | len >= 0 -> if isHead then - sendRsp conn ver s hs1 (RspBuilder mempty False) + sendRsp conn mfdc ver s hs1 (RspBuilder mempty False) else do lheader <- composeHeader ver s hs1 - connSendFile conn path beg len hook [lheader] - | otherwise -> do - sendRsp conn ver H.status416 +#ifdef WINDOWS + let fid = FileId path Nothing + hook' = hook +#else + (mfd, hook') <- case mfdc of + -- Windows or settingsFdCacheDuration is 0 + Nothing -> return (Nothing, hook) + Just fdc -> do + (fd, fresher) <- F.getFd fdc path + return (Just fd, hook >> fresher) + let fid = FileId path mfd +#endif + connSendFile conn fid beg len hook' [lheader] + | otherwise -> + sendRsp conn mfdc ver H.status416 (filter (\(k, _) -> k /= "content-length") hs1) (RspBuilder mempty True) where @@ -249,7 +272,7 @@ ---------------------------------------------------------------- -sendRsp conn ver s hs (RspBuilder body needsChunked) = do +sendRsp conn _ ver s hs (RspBuilder body needsChunked) = do header <- composeHeaderBuilder ver s hs needsChunked let hdrBdy | needsChunked = header <> chunkedTransferEncoding body @@ -261,7 +284,7 @@ ---------------------------------------------------------------- -sendRsp conn ver s hs (RspStream streamingBody needsChunked th) = do +sendRsp conn _ ver s hs (RspStream streamingBody needsChunked th) = do header <- composeHeaderBuilder ver s hs needsChunked (recv, finish) <- newBlazeRecv $ reuseBufferStrategy $ toBlazeBuffer (connWriteBuffer conn) (connBufferSize conn) @@ -284,7 +307,7 @@ ---------------------------------------------------------------- -sendRsp conn _ _ _ (RspRaw withApp src tickle) = +sendRsp conn _ _ _ _ (RspRaw withApp src tickle) = withApp recv send where recv = do @@ -357,13 +380,12 @@ ---------------------------------------------------------------- -hasBody :: H.Status -> Request -> Bool -hasBody s req = sc /= 204 - && sc /= 304 - && sc >= 200 +hasBody :: H.Status -> Bool +hasBody s = sc /= 204 + && sc /= 304 + && sc >= 200 where sc = H.statusCode s - method = requestMethod req ---------------------------------------------------------------- @@ -386,9 +408,9 @@ -- building with ShowS $ 'b' : 'y': 't' : 'e' : 's' : ' ' : (if beg > end then ('*':) else - (showInt beg) + showInt beg . ('-' :) - . (showInt end)) + . showInt end) ( '/' : showInt total "") diff --git a/Network/Wai/Handler/Warp/ResponseHeader.hs b/Network/Wai/Handler/Warp/ResponseHeader.hs --- a/Network/Wai/Handler/Warp/ResponseHeader.hs +++ b/Network/Wai/Handler/Warp/ResponseHeader.hs @@ -6,14 +6,14 @@ import Control.Monad import Data.ByteString (ByteString) import qualified Data.ByteString as S -import Data.ByteString.Internal (ByteString(..), create, memcpy) +import Data.ByteString.Internal (create) import qualified Data.CaseInsensitive as CI import Data.List (foldl') import Data.Word (Word8) -import Foreign.ForeignPtr import Foreign.Ptr import GHC.Storable import qualified Network.HTTP.Types as H +import Network.Wai.Handler.Warp.Buffer (copy) ---------------------------------------------------------------- @@ -27,12 +27,6 @@ fieldLength !l !(k,v) = l + S.length (CI.original k) + S.length v + 4 !slen = S.length $ H.statusMessage status -{-# INLINE copy #-} -copy :: Ptr Word8 -> ByteString -> IO (Ptr Word8) -copy !ptr (PS fp o l) = withForeignPtr fp $ \p -> do - memcpy ptr (p `plusPtr` o) (fromIntegral l) - return $! ptr `plusPtr` l - httpVer11 :: ByteString httpVer11 = "HTTP/1.1 " @@ -90,5 +84,3 @@ cr = 13 lf :: Word8 lf = 10 - - diff --git a/Network/Wai/Handler/Warp/Run.hs b/Network/Wai/Handler/Warp/Run.hs --- a/Network/Wai/Handler/Warp/Run.hs +++ b/Network/Wai/Handler/Warp/Run.hs @@ -25,6 +25,7 @@ import Network.Wai.Handler.Warp.Counter import qualified Network.Wai.Handler.Warp.Date as D import qualified Network.Wai.Handler.Warp.FdCache as F +import Network.Wai.Handler.Warp.HTTP2 import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.ReadInt import Network.Wai.Handler.Warp.Recv @@ -45,21 +46,21 @@ import Network.Socket (fdSocket) #endif --- | Default action value for 'Connection'. +-- | Creating 'Connection' for plain HTTP based on a given socket. socketConnection :: Socket -> IO Connection socketConnection s = do - readBuf <- allocateBuffer bufferSize + bufferPool <- newBufferPool writeBuf <- allocateBuffer bufferSize + let sendall = Sock.sendAll s return Connection { connSendMany = Sock.sendMany s - , connSendAll = Sock.sendAll s - , connSendFile = defaultSendFile s - , connClose = sClose s >> freeBuffer readBuf >> freeBuffer writeBuf - , connRecv = receive s readBuf bufferSize - , connReadBuffer = readBuf + , connSendAll = sendall + , connSendFile = sendFile s writeBuf bufferSize sendall + , connClose = sClose s >> freeBuffer writeBuf + , connRecv = receive s bufferPool + , connRecvBuf = receiveBuf s , connWriteBuffer = writeBuf , connBufferSize = bufferSize - , connSendFileOverride = Override s } #if __GLASGOW_HASKELL__ < 702 @@ -67,14 +68,14 @@ allowInterrupt = unblock $ return () #endif --- | Run an 'Application' on the given port. This calls 'runSettings' with --- 'defaultSettings'. +-- | Run an 'Application' on the given port. +-- This calls 'runSettings' with 'defaultSettings'. run :: Port -> Application -> IO () run p = runSettings defaultSettings { settingsPort = p } --- | Run an 'Application' on the port present in the @PORT@ environment variable --- --- Uses the 'Port' given when the variable is unset. +-- | Run an 'Application' on the port present in the @PORT@ +-- environment variable. Uses the 'Port' given when the variable is unset. +-- This calls 'runSettings' with 'defaultSettings'. -- -- Since 3.0.9 runEnv :: Port -> Application -> IO () @@ -90,6 +91,8 @@ _ -> fail $ "Invalid value in $PORT: " ++ sp -- | Run an 'Application' with the given 'Settings'. +-- This opens a listen socket on the port defined in 'Settings' and +-- calls 'runSettingsSocket'. runSettings :: Settings -> Application -> IO () runSettings set app = withSocketsDo $ bracket @@ -99,15 +102,17 @@ setSocketCloseOnExec socket runSettingsSocket set socket app) --- | Same as 'runSettings', but uses a user-supplied socket instead of opening --- one. This allows the user to provide, for example, Unix named socket, which +-- | This installs a shutdown handler for the given socket and +-- calls 'runSettingsConnection' with the default connection setup action +-- which handles plain (non-cipher) HTTP. +-- When the listen socket in the second argument is closed, all live +-- connections are gracefully shut down. +-- +-- The supplied socket can be a Unix named socket, which -- can be used when reverse HTTP proxying into your application. -- -- Note that the 'settingsPort' will still be passed to 'Application's via the -- 'serverPort' record. --- --- When the listen socket in the second argument is closed, all live --- connections are gracefully shut down. runSettingsSocket :: Settings -> Socket -> Application -> IO () runSettingsSocket set socket app = do settingsInstallShutdownHandler set closeListenSocket @@ -125,10 +130,13 @@ closeListenSocket = sClose socket --- | Allows you to provide a function which will return a 'Connection'. In --- cases where creating the @Connection@ can be expensive, this allows the --- expensive computations to be performed in a separate thread instead of the --- main server loop. +-- | The connection setup action would be expensive. A good example +-- is initialization of TLS. +-- So, this converts the connection setup action to the connection maker +-- which will be executed after forking a new worker thread. +-- Then this calls 'runSettingsConnectionMaker' with the connection maker. +-- This allows the expensive computations to be performed +-- in a separate worker thread instead of the main server loop. -- -- Since 1.3.5 runSettingsConnection :: Settings -> IO (Connection, SockAddr) -> Application -> IO () @@ -138,6 +146,8 @@ (conn, sa) <- getConn return (return conn, sa) +-- | This modifies the connection maker so that it returns 'TCP' for 'Transport' +-- (i.e. plain HTTP) then calls 'runSettingsConnectionMakerSecure'. runSettingsConnectionMaker :: Settings -> IO (IO Connection, SockAddr) -> Application -> IO () runSettingsConnectionMaker x y = runSettingsConnectionMakerSecure x (go y) @@ -146,8 +156,10 @@ ---------------------------------------------------------------- --- | Allows you to provide a function which will return a function --- which will return 'Connection'. +-- | The core run function which takes 'Settings', +-- a connection maker and 'Application'. +-- The connection maker can return a connection of either plain HTTP +-- or HTTP over TLS. -- -- Since 2.1.4 runSettingsConnectionMakerSecure :: Settings -> IO (IO (Connection, Transport), SockAddr) -> Application -> IO () @@ -168,11 +180,6 @@ T.stopManager f -onE :: Settings -> Maybe Request -> SomeException -> IO () -onE set mreq e = case fromException e of - Just (NotEnoughLines []) -> return () - _ -> settingsOnException set mreq e - -- Note that there is a thorough discussion of the exception safety of the -- following code at: https://github.com/yesodweb/wai/issues/146 -- @@ -198,7 +205,7 @@ -- First mask all exceptions in acceptLoop. This is necessary to -- ensure that no async exception is throw between the call to -- acceptNewConnection and the registering of connClose. - void $ mask_ $ acceptLoop + void $ mask_ acceptLoop gracefulShutdown counter where acceptLoop = do @@ -224,7 +231,7 @@ case ex of Right x -> return $ Just x Left e -> do - onE set Nothing $ toException e + settingsOnException set Nothing $ toException e if isFullErrorType (ioeGetErrorType e) then do -- "resource exhausted (Too many open files)" may -- happen by accept(). Wait a second hoping that @@ -257,21 +264,20 @@ -- We grab the connection before registering timeouts since the -- timeouts will be useless during connection creation, due to the -- fact that async exceptions are still masked. - bracket mkConn closeConn $ \(conn0, transport) -> + bracket mkConn closeConn $ \(conn, transport) -> -- We need to register a timeout handler for this thread, and -- cancel that handler as soon as we exit. bracket (T.registerKillThread tm) T.cancel $ \th -> - let ii = InternalInfo th fc dc - conn = setSendFile conn0 fc + let ii = InternalInfo th tm fc dc -- We now have fully registered a connection close handler -- in the case of all exceptions, so it is safe to one -- again allow async exceptions. in unmask . -- Call the user-supplied on exception code if any -- exceptions are thrown. - handle (onE set Nothing) . + handle (settingsOnException set Nothing) . -- Call the user-supplied code for connection open and close events bracket (onOpen addr) (onClose addr) $ \goingon -> @@ -293,13 +299,28 @@ -> Application -> IO () serveConnection conn ii origAddr transport settings app = do - istatus <- newIORef False - src <- mkSource (connSource conn th istatus) - addr <- getProxyProtocolAddr src - http1 addr istatus src `E.catch` \e -> do - sendErrorResponse addr istatus e - throwIO (e :: SomeException) - + -- fixme: Upgrading to HTTP/2 should be supported. + (h2,bs) <- if isHTTP2 transport then + return (True, "") + else do + bs0 <- connRecv conn + if S.length bs0 >= 4 && "PRI " `S.isPrefixOf` bs0 then + return (True, bs0) + else + return (False, bs0) + if h2 then do + recvN <- makeReceiveN bs (connRecv conn) (connRecvBuf conn) + -- fixme: origAddr + http2 conn ii origAddr transport settings recvN app + else do + istatus <- newIORef False + src <- mkSource (wrappedRecv conn th istatus) + writeIORef istatus True + leftoverSource src bs + addr <- getProxyProtocolAddr src + http1 addr istatus src `E.catch` \e -> do + sendErrorResponse addr istatus e + throwIO (e :: SomeException) where getProxyProtocolAddr src = case settingsProxyProtocol settings of @@ -361,8 +382,8 @@ `E.catch` \e -> do -- Call the user-supplied exception handlers, passing the request. sendErrorResponse addr istatus e - onE settings (Just req) e - -- Don't throw the error again to prevent calling onE twice. + settingsOnException settings (Just req) e + -- Don't throw the error again to prevent calling settingsOnException twice. return False when keepAlive $ http1 addr istatus src @@ -399,7 +420,7 @@ if not keepAlive then return False - else do + else -- If there is an unknown or large amount of data to still be read -- from the request body, simple drop this connection instead of -- reading it all in to satisfy a keep-alive request. @@ -449,8 +470,8 @@ | toRead' >= 0 -> loop toRead' | otherwise -> return False -connSource :: Connection -> T.Handle -> IORef Bool -> IO ByteString -connSource Connection { connRecv = recv } th istatus = do +wrappedRecv :: Connection -> T.Handle -> IORef Bool -> IO ByteString +wrappedRecv Connection { connRecv = recv } th istatus = do bs <- recv unless (S.null bs) $ do writeIORef istatus True @@ -467,8 +488,4 @@ #endif gracefulShutdown :: Counter -> IO () -gracefulShutdown counter = do - -- To avoid race condition, we just use threadDelay, not MVar. - threadDelay 10000000 - noConnections <- isZero counter - unless noConnections $ gracefulShutdown counter +gracefulShutdown counter = waitForZero counter diff --git a/Network/Wai/Handler/Warp/SendFile.hs b/Network/Wai/Handler/Warp/SendFile.hs --- a/Network/Wai/Handler/Warp/SendFile.hs +++ b/Network/Wai/Handler/Warp/SendFile.hs @@ -1,29 +1,151 @@ -{-# LANGUAGE CPP #-} +{-# LANGUAGE CPP, ForeignFunctionInterface #-} -module Network.Wai.Handler.Warp.SendFile where +module Network.Wai.Handler.Warp.SendFile ( + sendFile + , readSendFile + , packHeader -- for testing +#ifndef WINDOWS + , positionRead +#endif + ) where +import Control.Monad (void) import Data.ByteString (ByteString) -import Network.Sendfile +import qualified Data.ByteString as BS import Network.Socket (Socket) -import qualified Network.Wai.Handler.Warp.FdCache as F +import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.Types -defaultSendFile :: Socket -> FilePath -> Integer -> Integer -> IO () -> [ByteString] -> IO () -defaultSendFile s path off len act hdr = sendfileWithHeader s path (PartOfFile off len) act hdr +#ifdef WINDOWS +import Control.Monad (when) +import Data.ByteString.Internal (ByteString(..)) +import Foreign.ForeignPtr (newForeignPtr_) +import Foreign.Ptr (plusPtr) +import qualified System.IO as IO +#else +# if __GLASGOW_HASKELL__ < 709 +import Control.Applicative ((<$>)) +# endif +import Control.Exception +import Foreign.C.Types +import Foreign.Ptr (Ptr, castPtr, plusPtr) +import Network.Sendfile +import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd) +import System.Posix.Types +#endif +---------------------------------------------------------------- -#if SENDFILEFD -setSendFile :: Connection -> Maybe F.MutableFdCache -> Connection -setSendFile conn Nothing = conn -setSendFile conn (Just fdcs) = case connSendFileOverride conn of - NotOverride -> conn - Override s -> conn { connSendFile = sendFile fdcs s } +-- | Function to send a file based on sendfile() for Linux\/Mac\/FreeBSD. +-- This makes use of the file descriptor cache. +-- For other OSes, this is identical to 'readSendFile'. +-- +-- Since: 3.1.0 +sendFile :: Socket -> Buffer -> BufSize -> (ByteString -> IO ()) -> SendFile +#ifdef SENDFILEFD +sendFile s _ _ _ fid off len act hdr = case mfid of + -- settingsFdCacheDuration is 0 + Nothing -> sendfileWithHeader s path (PartOfFile off len) act hdr + Just fd -> sendfileFdWithHeader s fd (PartOfFile off len) act hdr + where + mfid = fileIdFd fid + path = fileIdPath fid +#else +sendFile _ = readSendFile +#endif -sendFile :: F.MutableFdCache -> Socket -> FilePath -> Integer -> Integer -> IO () -> [ByteString] -> IO () -sendFile fdcs s path off len act hdr = do - (fd, fresher) <- F.getFd fdcs path - sendfileFdWithHeader s fd (PartOfFile off len) (act>>fresher) hdr +---------------------------------------------------------------- + +packHeader :: Buffer -> BufSize -> (ByteString -> IO ()) + -> IO () -> [ByteString] + -> Int + -> IO Int +packHeader _ _ _ _ [] n = return n +packHeader buf siz send hook (bs:bss) n + | len < room = do + let dst = buf `plusPtr` n + void $ copy dst bs + packHeader buf siz send hook bss (n + len) + | otherwise = do + let dst = buf `plusPtr` n + (bs1, bs2) = BS.splitAt room bs + void $ copy dst bs1 + bufferIO buf siz send + hook + packHeader buf siz send hook (bs2:bss) 0 + where + len = BS.length bs + room = siz - n + +mini :: Int -> Integer -> Int +mini i n + | fromIntegral i < n = i + | otherwise = fromIntegral n + +-- | Function to send a file based on pread()\/send() for Unix. +-- This makes use of the file descriptor cache. +-- For Windows, this is emulated by 'Handle'. +-- +-- Since: 3.1.0 +#ifdef WINDOWS +readSendFile :: Buffer -> BufSize -> (ByteString -> IO ()) -> SendFile +readSendFile buf siz send fid off0 len0 hook headers = do + hn <- packHeader buf siz send hook headers 0 + let room = siz - hn + buf' = buf `plusPtr` hn + IO.withBinaryFile path IO.ReadMode $ \h -> do + IO.hSeek h IO.AbsoluteSeek off0 + n <- IO.hGetBufSome h buf' (mini room len0) + bufferIO buf (hn + n) send + hook + let n' = fromIntegral n + fptr <- newForeignPtr_ buf + loop h fptr (len0 - n') + where + path = fileIdPath fid + loop h fptr len + | len <= 0 = return () + | otherwise = do + n <- IO.hGetBufSome h buf (mini siz len) + when (n /= 0) $ do + let bs = PS fptr 0 n + n' = fromIntegral n + send bs + hook + loop h fptr (len - n') #else -setSendFile :: Connection -> Maybe F.MutableFdCache -> Connection -setSendFile conn _ = conn +readSendFile :: Buffer -> BufSize -> (ByteString -> IO ()) -> SendFile +readSendFile buf siz send fid off0 len0 hook headers = + bracket setup teardown $ \fd -> do + hn <- packHeader buf siz send hook headers 0 + let room = siz - hn + buf' = buf `plusPtr` hn + n <- positionRead fd buf' (mini room len0) off0 + bufferIO buf (hn + n) send + hook + let n' = fromIntegral n + loop fd (len0 - n') (off0 + n') + where + path = fileIdPath fid + setup = case fileIdFd fid of + Just fd -> return fd + Nothing -> openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True} + teardown fd = case fileIdFd fid of + Just _ -> return () + Nothing -> closeFd fd + loop fd len off + | len <= 0 = return () + | otherwise = do + n <- positionRead fd buf (mini siz len) off + bufferIO buf n send + let n' = fromIntegral n + hook + loop fd (len - n') (off + n') + +positionRead :: Fd -> Buffer -> BufSize -> Integer -> IO Int +positionRead fd buf siz off = + fromIntegral <$> c_pread fd (castPtr buf) (fromIntegral siz) (fromIntegral off) + +foreign import ccall unsafe "pread" + c_pread :: Fd -> Ptr CChar -> ByteCount -> FileOffset -> IO CSsize #endif diff --git a/Network/Wai/Handler/Warp/Settings.hs b/Network/Wai/Handler/Warp/Settings.hs --- a/Network/Wai/Handler/Warp/Settings.hs +++ b/Network/Wai/Handler/Warp/Settings.hs @@ -1,29 +1,32 @@ {-# LANGUAGE OverloadedStrings, ScopedTypeVariables, ViewPatterns #-} {-# LANGUAGE PatternGuards, RankNTypes #-} -{-# LANGUAGE ImpredicativeTypes #-} +{-# LANGUAGE ImpredicativeTypes, CPP #-} module Network.Wai.Handler.Warp.Settings where +import Blaze.ByteString.Builder (copyByteString) +import Blaze.ByteString.Builder.Char.Utf8 (fromShow) +import Control.Concurrent (forkIOWithUnmask) import Control.Exception import Control.Monad (when, void) -import Control.Concurrent (forkIOWithUnmask) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as S8 +#if __GLASGOW_HASKELL__ < 709 +import Data.Monoid (mappend) +#endif +import Data.Streaming.Network (HostPreference) import qualified Data.Text as T import qualified Data.Text.IO as TIO -import Data.Streaming.Network (HostPreference) +import Data.Version (showVersion) import GHC.IO.Exception (IOErrorType(..)) import qualified Network.HTTP.Types as H import Network.Socket (SockAddr) import Network.Wai import Network.Wai.Handler.Warp.Timeout import Network.Wai.Handler.Warp.Types +import qualified Paths_warp import System.IO (stderr) import System.IO.Error (ioeGetErrorType) -import qualified Data.Text.Lazy as TL -import qualified Data.Text.Lazy.Encoding as TLE -import Data.ByteString (ByteString) -import qualified Data.ByteString.Char8 as S8 -import Data.Version (showVersion) -import qualified Paths_warp -- | Various Warp server settings. This is purposely kept as an abstract data -- type so that new settings can be added without breaking backwards @@ -32,7 +35,7 @@ -- -- > setTimeout 20 defaultSettings data Settings = Settings - { settingsPort :: Int -- ^ Port to listen on. Default value: 3000 + { settingsPort :: Port -- ^ Port to listen on. Default value: 3000 , settingsHost :: HostPreference -- ^ Default value: HostIPv4 , settingsOnException :: Maybe Request -> SomeException -> IO () -- ^ What to do with exceptions thrown by either the application or server. Default: ignore server-generated exceptions (see 'InvalidRequest') and print application-generated applications to stderr. , settingsOnExceptionResponse :: SomeException -> Response @@ -102,8 +105,8 @@ defaultSettings = Settings { settingsPort = 3000 , settingsHost = "*4" - , settingsOnException = defaultExceptionHandler - , settingsOnExceptionResponse = defaultExceptionResponse + , settingsOnException = defaultOnException + , settingsOnExceptionResponse = defaultOnExceptionResponse , settingsOnOpen = const $ return True , settingsOnClose = const $ return () , settingsTimeout = 30 @@ -118,7 +121,7 @@ , settingsProxyProtocol = ProxyProtocolNone } --- | Apply the logic provided by 'defaultExceptionHandler' to determine if an +-- | Apply the logic provided by 'defaultOnException' to determine if an -- exception should be shown or not. The goal is to hide exceptions which occur -- under the normal course of the web server running. -- @@ -132,26 +135,27 @@ | Just TimeoutThread <- fromException se = False | otherwise = True -defaultExceptionHandler :: Maybe Request -> SomeException -> IO () -defaultExceptionHandler _ e = +-- | Printing an exception to standard error +-- if `defaultShouldDisplayException` returns `True`. +-- +-- Since: 3.1.0 +defaultOnException :: Maybe Request -> SomeException -> IO () +defaultOnException _ e = when (defaultShouldDisplayException e) $ TIO.hPutStrLn stderr $ T.pack $ show e -defaultExceptionResponse :: SomeException -> Response -defaultExceptionResponse _ = responseLBS H.internalServerError500 [(H.hContentType, "text/plain; charset=utf-8")] "Something went wrong" +-- | Sending 400 for bad requests. Sending 500 for internal server errors. +-- +-- Since: 3.1.0 +defaultOnExceptionResponse :: SomeException -> Response +defaultOnExceptionResponse e + | Just (_ :: InvalidRequest) <- fromException e = responseLBS H.badRequest400 [(H.hContentType, "text/plain; charset=utf-8")] "Bad Request" + | otherwise = responseLBS H.internalServerError500 [(H.hContentType, "text/plain; charset=utf-8")] "Something went wrong" --- | Default implementation of 'settingsOnExceptionResponse' for the debugging purpose. 500, text/plain, a showed exception. +-- | Exception handler for the debugging purpose. +-- 500, text/plain, a showed exception. +-- +-- Since: 2.0.3.2 exceptionResponseForDebug :: SomeException -> Response -exceptionResponseForDebug e = responseLBS H.internalServerError500 [(H.hContentType, "text/plain; charset=utf-8")] (TLE.encodeUtf8 $ TL.pack $ "Exception: " ++ show e) - -{-# DEPRECATED settingsPort "Use setPort instead" #-} -{-# DEPRECATED settingsHost "Use setHost instead" #-} -{-# DEPRECATED settingsOnException "Use setOnException instead" #-} -{-# DEPRECATED settingsOnExceptionResponse "Use setOnExceptionResponse instead" #-} -{-# DEPRECATED settingsOnOpen "Use setOnOpen instead" #-} -{-# DEPRECATED settingsOnClose "Use setOnClose instead" #-} -{-# DEPRECATED settingsTimeout "Use setTimeout instead" #-} -{-# DEPRECATED settingsManager "Use setManager instead" #-} -{-# DEPRECATED settingsFdCacheDuration "Use setFdCacheDuration instead" #-} -{-# DEPRECATED settingsBeforeMainLoop "Use setBeforeMainLoop instead" #-} -{-# DEPRECATED settingsNoParsePath "Use setNoParsePath instead" #-} +exceptionResponseForDebug e = responseBuilder H.internalServerError500 [(H.hContentType, "text/plain; charset=utf-8")] + $ copyByteString "Exception: " `mappend` fromShow e diff --git a/Network/Wai/Handler/Warp/Timeout.hs b/Network/Wai/Handler/Warp/Timeout.hs --- a/Network/Wai/Handler/Warp/Timeout.hs +++ b/Network/Wai/Handler/Warp/Timeout.hs @@ -3,42 +3,24 @@ {-# LANGUAGE MagicHash #-} {-# LANGUAGE UnboxedTuples #-} -- for GHC 7.4 or earlier --- | --- --- In order to provide slowloris protection, Warp provides timeout handlers. We --- follow these rules: --- --- * A timeout is created when a connection is opened. --- --- * When all request headers are read, the timeout is tickled. --- --- * Every time at least 2048 bytes of the request body are read, the timeout --- is tickled. --- --- * The timeout is paused while executing user code. This will apply to both --- the application itself, and a ResponseSource response. The timeout is --- resumed as soon as we return from user code. --- --- * Every time data is successfully sent to the client, the timeout is tickled. - module Network.Wai.Handler.Warp.Timeout ( - -- * Types + -- ** Types Manager , TimeoutAction , Handle - -- * Manager + -- ** Manager , initialize , stopManager , withManager - -- * Registration + -- ** Registration , register , registerKillThread - -- * Control + -- ** Control , tickle , cancel , pause , resume - -- * Exceptions + -- ** Exceptions , TimeoutThread (..) ) where diff --git a/Network/Wai/Handler/Warp/Types.hs b/Network/Wai/Handler/Warp/Types.hs --- a/Network/Wai/Handler/Warp/Types.hs +++ b/Network/Wai/Handler/Warp/Types.hs @@ -9,14 +9,17 @@ import qualified Data.ByteString as S import Data.IORef (IORef, readIORef, writeIORef, newIORef) import Data.Typeable (Typeable) -import Data.Word (Word16) +import Data.Word (Word16, Word8) +import Foreign.Ptr (Ptr) import Network.HTTP.Types.Header -import Network.Socket (Socket) -import Network.Wai.Handler.Warp.Buffer (Buffer,BufSize) import qualified Network.Wai.Handler.Warp.Date as D import qualified Network.Wai.Handler.Warp.FdCache as F import qualified Network.Wai.Handler.Warp.Timeout as T +#ifndef WINDOWS +import System.Posix.Types (Fd) +#endif + ---------------------------------------------------------------- -- | TCP port number. @@ -64,25 +67,61 @@ ---------------------------------------------------------------- --- | Whether or not 'ConnSendFileOverride' in 'Connection' can be --- overridden. This is a kind of hack to keep the signature of --- 'Connection' clean. -data ConnSendFileOverride = NotOverride -- ^ Don't override - | Override Socket -- ^ Override with this 'Socket' +#ifdef WINDOWS +type Fd = () +#endif ----------------------------------------------------------------- +-- | Data type to abstract file identifiers. +-- On Unix, a file descriptor would be specified to make use of +-- the file descriptor cache. +-- +-- Since: 3.1.0 +data FileId = FileId { + fileIdPath :: FilePath + , fileIdFd :: Maybe Fd + } +-- | fileid, offset, length, hook action, HTTP headers +-- +-- Since: 3.1.0 +type SendFile = FileId -> Integer -> Integer -> IO () -> [ByteString] -> IO () + +-- | Type for read buffer pool +type BufferPool = IORef ByteString + +-- | Type for buffer +type Buffer = Ptr Word8 + +-- | Type for buffer size +type BufSize = Int + +-- | Type for the action to receive input data +type Recv = IO ByteString + +-- | Type for the action to receive input data with a buffer. +-- The result boolean indicates whether or not the buffer is fully filled. +type RecvBuf = Buffer -> BufSize -> IO Bool + -- | Data type to manipulate IO actions for connections. -data Connection = Connection - { connSendMany :: [ByteString] -> IO () - , connSendAll :: ByteString -> IO () - , connSendFile :: FilePath -> Integer -> Integer -> IO () -> [ByteString] -> IO () -- ^ filepath, offset, length, hook action, HTTP headers - , connClose :: IO () - , connRecv :: IO ByteString - , connReadBuffer :: Buffer - , connWriteBuffer :: Buffer - , connBufferSize :: BufSize - , connSendFileOverride :: ConnSendFileOverride +-- This is used to abstract IO actions for plain HTTP and HTTP over TLS. +data Connection = Connection { + -- | This is not used at this moment. + connSendMany :: [ByteString] -> IO () + -- | The sending function. + , connSendAll :: ByteString -> IO () + -- | The sending function for files in HTTP/1.1. + , connSendFile :: SendFile + -- | The connection closing function. + , connClose :: IO () + -- | The connection receiving function. This returns "" for EOF. + , connRecv :: Recv + -- | The connection receiving function. This tries to fill the buffer. + -- This returns when the buffer is filled or reaches EOF. + , connRecvBuf :: RecvBuf + -- | The write buffer. + , connWriteBuffer :: Buffer + -- | The size of the write buffer. + , connBufferSize :: BufSize } ---------------------------------------------------------------- @@ -90,12 +129,14 @@ -- | Internal information. data InternalInfo = InternalInfo { threadHandle :: T.Handle + , timeoutManager :: T.Manager , fdCacher :: Maybe F.MutableFdCache , dateCacher :: D.DateCache } ---------------------------------------------------------------- +-- | Type for input streaming. data Source = Source !(IORef ByteString) !(IO ByteString) mkSource :: IO ByteString -> IO Source diff --git a/test/ExceptionSpec.hs b/test/ExceptionSpec.hs --- a/test/ExceptionSpec.hs +++ b/test/ExceptionSpec.hs @@ -1,8 +1,10 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings, CPP #-} module ExceptionSpec (main, spec) where +#if __GLASGOW_HASKELL__ < 709 import Control.Applicative +#endif import Control.Monad import Network.HTTP import Network.Stream diff --git a/test/FdCacheSpec.hs b/test/FdCacheSpec.hs --- a/test/FdCacheSpec.hs +++ b/test/FdCacheSpec.hs @@ -3,7 +3,7 @@ module FdCacheSpec where import Test.Hspec -#ifdef SENDFILEFD +#ifndef WINDOWS import Data.IORef import Network.Wai.Handler.Warp.FdCache import System.Posix.IO (fdRead) diff --git a/test/SendFileSpec.hs b/test/SendFileSpec.hs new file mode 100644 --- /dev/null +++ b/test/SendFileSpec.hs @@ -0,0 +1,116 @@ +{-# LANGUAGE OverloadedStrings #-} + +module SendFileSpec where + +import Control.Exception +import Control.Monad (when) +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Network.Wai.Handler.Warp.Buffer +import Network.Wai.Handler.Warp.SendFile +import Network.Wai.Handler.Warp.Types +import System.Directory +import System.Exit +import qualified System.IO as IO +import System.Process (system) +import Test.Hspec + +main :: IO () +main = hspec spec + +spec :: Spec +spec = do + describe "packHeader" $ do + it "returns how much the buffer is consumed (1)" $ + tryPackHeader 10 ["foo"] `shouldReturn` 3 + it "returns how much the buffer is consumed (2)" $ + tryPackHeader 10 ["foo", "bar"] `shouldReturn` 6 + it "returns how much the buffer is consumed (3)" $ + tryPackHeader 10 ["0123456789"] `shouldReturn` 0 + it "returns how much the buffer is consumed (4)" $ + tryPackHeader 10 ["01234", "56789"] `shouldReturn` 0 + it "returns how much the buffer is consumed (5)" $ + tryPackHeader 10 ["01234567890", "12"] `shouldReturn` 3 + it "returns how much the buffer is consumed (6)" $ + tryPackHeader 10 ["012345678901234567890123456789012", "34"] `shouldReturn` 5 + + it "sends headers correctly (1)" $ + tryPackHeader2 10 ["foo"] "" `shouldReturn` True + it "sends headers correctly (2)" $ + tryPackHeader2 10 ["foo", "bar"] "" `shouldReturn` True + it "sends headers correctly (3)" $ + tryPackHeader2 10 ["0123456789"] "0123456789" `shouldReturn` True + it "sends headers correctly (4)" $ + tryPackHeader2 10 ["01234", "56789"] "0123456789" `shouldReturn` True + it "sends headers correctly (5)" $ + tryPackHeader2 10 ["01234567890", "12"] "0123456789" `shouldReturn` True + it "sends headers correctly (6)" $ + tryPackHeader2 10 ["012345678901234567890123456789012", "34"] "012345678901234567890123456789" `shouldReturn` True + + describe "readSendFile" $ do + it "sends a file correctly (1)" $ + tryReadSendFile 10 0 1474 ["foo"] `shouldReturn` ExitSuccess + it "sends a file correctly (2)" $ + tryReadSendFile 10 0 1474 ["012345678", "901234"] `shouldReturn` ExitSuccess + it "sends a file correctly (3)" $ + tryReadSendFile 10 20 100 ["012345678", "901234"] `shouldReturn` ExitSuccess + +tryPackHeader :: Int -> [ByteString] -> IO Int +tryPackHeader siz hdrs = bracket (allocateBuffer siz) freeBuffer $ \buf -> + packHeader buf siz send hook hdrs 0 + where + send _ = return () + hook = return () + +tryPackHeader2 :: Int -> [ByteString] -> ByteString -> IO Bool +tryPackHeader2 siz hdrs ans = bracket setup teardown $ \buf -> do + _ <- packHeader buf siz send hook hdrs 0 + checkFile outputFile ans + where + setup = allocateBuffer siz + teardown buf = freeBuffer buf >> removeFileIfExists outputFile + outputFile = "tempfile" + send = BS.appendFile outputFile + hook = return () + +tryReadSendFile :: Int -> Integer -> Integer -> [ByteString] -> IO ExitCode +tryReadSendFile siz off len hdrs = bracket setup teardown $ \buf -> do + mapM_ (BS.appendFile expectedFile) hdrs + copyfile inputFile expectedFile off len + readSendFile buf siz send fid off len hook hdrs + compareFiles expectedFile outputFile + where + hook = return () + setup = allocateBuffer siz + teardown buf = do + freeBuffer buf + removeFileIfExists outputFile + removeFileIfExists expectedFile + inputFile = "test/inputFile" + outputFile = "outputFile" + expectedFile = "expectedFile" + fid = FileId inputFile Nothing + send = BS.appendFile outputFile + +checkFile :: FilePath -> ByteString -> IO Bool +checkFile path bs = do + exist <- doesFileExist path + if exist then do + bs' <- BS.readFile path + return $ bs == bs' + else + return $ bs == "" + +compareFiles :: FilePath -> FilePath -> IO ExitCode +compareFiles file1 file2 = system $ "cmp -s " ++ file1 ++ " " ++ file2 + +copyfile :: FilePath -> FilePath -> Integer -> Integer -> IO () +copyfile src dst off len = + IO.withBinaryFile src IO.ReadMode $ \h -> do + IO.hSeek h IO.AbsoluteSeek off + BS.hGet h (fromIntegral len) >>= BS.appendFile dst + +removeFileIfExists :: FilePath -> IO () +removeFileIfExists file = do + exist <- doesFileExist file + when exist $ removeFile file diff --git a/test/doctests.hs b/test/doctests.hs --- a/test/doctests.hs +++ b/test/doctests.hs @@ -1,11 +1,4 @@ -module Main where - import Test.DocTest main :: IO () -main = doctest [ - "-idist/build/autogen/" - , "-optP-include" - , "-optPdist/build/autogen/cabal_macros.h" - , "Network/Wai/Handler/Warp.hs" - ] +main = doctest ["Network"] diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 3.0.13.1 +Version: 3.1.0 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE @@ -10,7 +10,11 @@ Build-Type: Simple Cabal-Version: >=1.8 Stability: Stable -description: API docs and the README are available at . +description: HTTP\/1.0, HTTP\/1.1 and HTTP\/2 are supported. + For HTTP\/2, Warp supports direct and ALPN (in TLS) + but not upgrade. + API docs and the README are available at + . extra-source-files: attic/hex ChangeLog.md README.md @@ -37,15 +41,19 @@ , blaze-builder >= 0.3.3 && < 0.5 , bytestring >= 0.9.1.4 , case-insensitive >= 0.2 + , containers , ghc-prim , http-types >= 0.8.5 , iproute >= 1.3.1 + , http2 >= 1.0.2 , simple-sendfile >= 0.2.7 && < 0.3 , unix-compat >= 0.2 , wai >= 3.0 && < 3.1 , text , streaming-commons >= 0.1.10 , vault >= 0.3 + , stm >= 2.3 + , word8 if flag(network-bytestring) Build-Depends: network >= 2.2.1.5 && < 2.2.3 , network-bytestring >= 0.1.3 && < 0.1.4 @@ -57,13 +65,21 @@ else Build-Depends: bytestring >= 0.10.2.0 Exposed-modules: Network.Wai.Handler.Warp - Network.Wai.Handler.Warp.Buffer - Network.Wai.Handler.Warp.Timeout Network.Wai.Handler.Warp.Internal - Other-modules: Network.Wai.Handler.Warp.Conduit + Other-modules: Network.Wai.Handler.Warp.Buffer + Network.Wai.Handler.Warp.Conduit Network.Wai.Handler.Warp.Counter Network.Wai.Handler.Warp.Date Network.Wai.Handler.Warp.FdCache + Network.Wai.Handler.Warp.HTTP2 + Network.Wai.Handler.Warp.HTTP2.EncodeFrame + Network.Wai.Handler.Warp.HTTP2.HPACK + Network.Wai.Handler.Warp.HTTP2.Manager + Network.Wai.Handler.Warp.HTTP2.Receiver + Network.Wai.Handler.Warp.HTTP2.Request + Network.Wai.Handler.Warp.HTTP2.Sender + Network.Wai.Handler.Warp.HTTP2.Types + Network.Wai.Handler.Warp.HTTP2.Worker Network.Wai.Handler.Warp.Header Network.Wai.Handler.Warp.IO Network.Wai.Handler.Warp.IORef @@ -76,6 +92,7 @@ Network.Wai.Handler.Warp.Run Network.Wai.Handler.Warp.SendFile Network.Wai.Handler.Warp.Settings + Network.Wai.Handler.Warp.Timeout Network.Wai.Handler.Warp.Types Network.Wai.Handler.Warp.Windows Paths_warp @@ -101,7 +118,7 @@ Ghc-Options: -threaded -Wall Main-Is: doctests.hs Build-Depends: base - , doctest >= 0.9.3 + , doctest >= 0.10.1 Test-Suite spec Main-Is: Spec.hs @@ -114,6 +131,7 @@ ResponseHeaderSpec ResponseSpec RunSpec + SendFileSpec Hs-Source-Dirs: test, . Type: exitcode-stdio-1.0 @@ -143,6 +161,12 @@ , streaming-commons >= 0.1.10 , async , vault + , stm >= 2.3 + , directory + , process + , containers + , http2 >= 1.0.2 + , word8 if flag(use-bytestring-builder) Build-Depends: bytestring < 0.10.2.0 , bytestring-builder