2339bb5 · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

List-based timeouts

2339bb5Michael Snoyman16 years ago

Network/Wai/Handler/Warp.hs+61−47

compare with master

unifiedsplit

@@ -48,14 +48,12 @@ import qualified Data.ByteString as S

48import qualified Data.ByteString.Unsafe as SU

49import qualified Data.ByteString.Char8 as B

50import qualified Data.ByteString.Lazy as L

51−import Network

52− ( listenOn, sClose, PortID(PortNumber), Socket

51+import Network (listenOn, sClose, PortID(PortNumber), Socket)

53− , withSocketsDo)

54import Network.Socket

52import Network.Socket

55 ( accept, SockAddr

53 ( accept, SockAddr

56 )

54 )

57import qualified Network.Socket.ByteString as Sock

55import qualified Network.Socket.ByteString as Sock

58−import Control.Exception (bracket, finally, Exception, SomeException, catch, throwIO)

56+import Control.Exception (bracket, finally, Exception, SomeException, catch)

59import Control.Concurrent (forkIO)

57import Control.Concurrent (forkIO)

60import Data.Maybe (fromMaybe)

58import Data.Maybe (fromMaybe)

@@ -73,17 +71,17 @@ import Blaze.ByteString.Builder

73import Blaze.ByteString.Builder.Char8 (fromChar, fromShow)

71import Blaze.ByteString.Builder.Char8 (fromChar, fromShow)

74import Data.Monoid (mappend, mconcat)

72import Data.Monoid (mappend, mconcat)

75import Network.Socket.SendFile (sendFile)

73import Network.Socket.SendFile (sendFile)

76−import Network.Socket.Enumerator (iterSocket)

78import Control.Monad.IO.Class (liftIO)

75import Control.Monad.IO.Class (liftIO)

79−import System.Timeout (timeout)

76+import qualified Timeout as T

80import Data.Word (Word8)

77import Data.Word (Word8)

81import Data.List (foldl')

78import Data.List (foldl')

79+import Control.Monad (forever)

83#if WINDOWS

81#if WINDOWS

84import Control.Concurrent (threadDelay)

82import Control.Concurrent (threadDelay)

85import qualified Control.Concurrent.MVar as MV

83import qualified Control.Concurrent.MVar as MV

86−import Control.Monad (forever)

84+import Network.Socket (withSocketsDo)

87#endif

85#endif

89-- | Run an 'Application' on the given port, ignoring all exceptions.

87-- | Run an 'Application' on the given port, ignoring all exceptions.

@@ -105,7 +103,7 @@ runEx onE port app = withSocketsDo $ do

105 serveConnections onE port app s)

103 serveConnections onE port app s)

106 forever (threadDelay maxBound) `finally` clean

104 forever (threadDelay maxBound) `finally` clean

107#else

105#else

108−runEx onE port = withSocketsDo . -- FIXME should this be called by client user instead?

106+runEx onE port =

109 bracket

107 bracket

110 (listenOn $ PortNumber $ fromIntegral port)

108 (listenOn $ PortNumber $ fromIntegral port)

111 sClose .

109 sClose .

@@ -122,24 +120,32 @@ type Port = Int

122serveConnections :: (SomeException -> IO ())

120serveConnections :: (SomeException -> IO ())

123 -> Port -> Application -> Socket -> IO ()

121 -> Port -> Application -> Socket -> IO ()

124serveConnections onE port app socket = do

122serveConnections onE port app socket = do

123+ tm <- T.initialize $ timeout * 1000000

124+ forever $ do

125− (conn, sa) <- accept socket

125+ (conn, sa) <- accept socket

126− _ <- forkIO $ serveConnection onE port app conn sa

126+ _ <- forkIO $ do

127− serveConnections onE port app socket

127+ th <- T.register tm $ sClose conn

128−

128+ serveConnection th onE port app conn sa

129+ return ()

130+

131+serveConnection :: T.Handle

129−serveConnection :: (SomeException -> IO ())

132+ -> (SomeException -> IO ())

130 -> Port -> Application -> Socket -> SockAddr -> IO ()

133 -> Port -> Application -> Socket -> SockAddr -> IO ()

131−serveConnection onException port app conn remoteHost' = do

134+serveConnection th onException port app conn remoteHost' = do

132 catch

135 catch

133 (finally

136 (finally

134 (E.run_ $ fromClient $$ serveConnection')

137 (E.run_ $ fromClient $$ serveConnection')

135 (sClose conn))

138 (sClose conn))

136 onException

139 onException

137 where

140 where

138− fromClient = enumSocket bytesPerRead conn

141+ fromClient = enumSocket th bytesPerRead conn

139 serveConnection' = do

142 serveConnection' = do

140 (enumeratee, env) <- parseRequest port remoteHost'

143 (enumeratee, env) <- parseRequest port remoteHost'

144+ -- Let the application run for as long as it wants

145+ --FIXME liftIO $ T.pause th

141 res <- E.joinI $ enumeratee $$ app env

146 res <- E.joinI $ enumeratee $$ app env

147+ --liftIO $ T.resume th

142− keepAlive <- liftIO $ sendResponse env (httpVersion env) conn res

148+ keepAlive <- liftIO $ sendResponse th env (httpVersion env) conn res

143 if keepAlive then serveConnection' else return ()

149 if keepAlive then serveConnection' else return ()

144

150

145parseRequest :: Port -> SockAddr -> E.Iteratee S.ByteString IO (E.Enumeratee ByteString ByteString IO a, Request)

151parseRequest :: Port -> SockAddr -> E.Iteratee S.ByteString IO (E.Enumeratee ByteString ByteString IO a, Request)

@@ -148,11 +154,11 @@ parseRequest port remoteHost' = do

148 parseRequest' port headers' remoteHost'

154 parseRequest' port headers' remoteHost'

149

155

150-- FIXME come up with good values here

156-- FIXME come up with good values here

151−maxHeaders, maxHeaderLength, bytesPerRead, readTimeout :: Int

157+maxHeaders, maxHeaderLength, bytesPerRead, timeout :: Int

152maxHeaders = 30

158maxHeaders = 30

153maxHeaderLength = 1024

159maxHeaderLength = 1024

154bytesPerRead = 4096

160bytesPerRead = 4096

155−readTimeout = 3000000

161+timeout = 1 -- seconds

156

162

157data InvalidRequest =

163data InvalidRequest =

158 NotEnoughLines [String]

164 NotEnoughLines [String]

@@ -170,7 +176,7 @@ parseRequest' :: Port

170 -> [ByteString]

176 -> [ByteString]

171 -> SockAddr

177 -> SockAddr

172 -> E.Iteratee S.ByteString IO (E.Enumeratee S.ByteString S.ByteString IO a, Request)

178 -> E.Iteratee S.ByteString IO (E.Enumeratee S.ByteString S.ByteString IO a, Request)

173−parseRequest' _ [] _ = throwError $ NotEnoughLines []

179+parseRequest' _ [] _ = E.throwError $ NotEnoughLines []

174parseRequest' port (firstLine:otherLines) remoteHost' = do

180parseRequest' port (firstLine:otherLines) remoteHost' = do

175 (method, rpath', gets, httpversion) <- parseFirst firstLine

181 (method, rpath', gets, httpversion) <- parseFirst firstLine

176 let rpath =

182 let rpath =

@@ -217,13 +223,13 @@ parseFirst s = do

217 (method, query, http') <-

223 (method, query, http') <-

218 case pieces of

224 case pieces of

219 [x, y, z] -> return (x, y, z)

225 [x, y, z] -> return (x, y, z)

220− _ -> throwError $ BadFirstLine $ B.unpack s

226+ _ -> E.throwError $ BadFirstLine $ B.unpack s

221 let (hfirst, hsecond) = B.splitAt 5 http'

227 let (hfirst, hsecond) = B.splitAt 5 http'

222 if (hfirst == "HTTP/")

228 if (hfirst == "HTTP/")

223 then

229 then

224 let (rpath, qstring) = B.break (== '?') query

230 let (rpath, qstring) = B.break (== '?') query

225 in return (method, rpath, qstring, hsecond)

231 in return (method, rpath, qstring, hsecond)

226− else throwError NonHttp

232+ else E.throwError NonHttp

227{-# INLINE parseFirst #-} -- FIXME is this inline necessary? the function is only called from one place and not exported

233{-# INLINE parseFirst #-} -- FIXME is this inline necessary? the function is only called from one place and not exported

228

234

229httpBuilder, spaceBuilder, newlineBuilder, transferEncodingBuilder

235httpBuilder, spaceBuilder, newlineBuilder, transferEncodingBuilder

@@ -262,15 +268,16 @@ isChunked = (==) http11

262hasBody :: Status -> Request -> Bool

268hasBody :: Status -> Request -> Bool

263hasBody s req = s /= (Status 204 "") && requestMethod req /= "HEAD"

269hasBody s req = s /= (Status 204 "") && requestMethod req /= "HEAD"

264

270

271+sendResponse :: T.Handle

265−sendResponse :: Request -> HttpVersion -> Socket -> Response -> IO Bool

272+ -> Request -> HttpVersion -> Socket -> Response -> IO Bool

266−sendResponse req hv socket (ResponseFile s hs fp) = do

273+sendResponse _FIXMEth req hv socket (ResponseFile s hs fp) = do

267 Sock.sendMany socket $ L.toChunks $ toLazyByteString $ headers hv s hs False

274 Sock.sendMany socket $ L.toChunks $ toLazyByteString $ headers hv s hs False

268 if hasBody s req

275 if hasBody s req

269 then do

276 then do

270− sendFile socket fp

277+ sendFile socket fp -- FIXME tickle the timeout here

271 return $ lookup "content-length" hs /= Nothing

278 return $ lookup "content-length" hs /= Nothing

272 else return True

279 else return True

273−sendResponse req hv socket (ResponseBuilder s hs b)

280+sendResponse _FIXMEth req hv socket (ResponseBuilder s hs b)

274 | hasBody s req = do

281 | hasBody s req = do

275 toByteStringIO (Sock.sendAll socket) b'

282 toByteStringIO (Sock.sendAll socket) b'

276 return isKeepAlive

283 return isKeepAlive

@@ -291,7 +298,7 @@ sendResponse req hv socket (ResponseBuilder s hs b)

291 hasLength = lookup "content-length" hs /= Nothing

298 hasLength = lookup "content-length" hs /= Nothing

292 isChunked' = isChunked hv && not hasLength

299 isChunked' = isChunked hv && not hasLength

293 isKeepAlive = isChunked' || hasLength

300 isKeepAlive = isChunked' || hasLength

294−sendResponse req hv socket (ResponseEnumerator res) =

301+sendResponse th req hv socket (ResponseEnumerator res) =

295 res go

302 res go

296 where

303 where

297 -- FIXME perhaps alloca a buffer per thread and reuse that in all functiosn below. Should lessen greatly the GC burden (I hope)

304 -- FIXME perhaps alloca a buffer per thread and reuse that in all functiosn below. Should lessen greatly the GC burden (I hope)

@@ -305,7 +312,7 @@ sendResponse req hv socket (ResponseEnumerator res) =

305 chunk'

312 chunk'

306 $ E.enumList 1 [headers hv s hs isChunked']

313 $ E.enumList 1 [headers hv s hs isChunked']

307 $$ E.joinI $ builderToByteString -- FIXME unsafeBuilderToByteString

314 $$ E.joinI $ builderToByteString -- FIXME unsafeBuilderToByteString

308− $$ (iterSocket socket >> return isKeepAlive)

315+ $$ (iterSocket th socket >> return isKeepAlive)

309 where

316 where

310 hasLength = lookup "content-length" hs /= Nothing

317 hasLength = lookup "content-length" hs /= Nothing

311 isChunked' = isChunked hv && not hasLength

318 isChunked' = isChunked hv && not hasLength

@@ -331,24 +338,20 @@ parseHeaderNoAttr s =

331 else rest

338 else rest

332 in (mkCIByteString k, rest')

339 in (mkCIByteString k, rest')

333

340

334−-- FIXME when we switch to Jeremy's timeout code, we can probably start using

335−-- network-enumerator's enumSocket

336−enumSocket :: Int -> Socket -> E.Enumerator ByteString IO a

341+enumSocket :: T.Handle -> Int -> Socket -> E.Enumerator ByteString IO a

337−enumSocket len socket (E.Continue k) = do

342+enumSocket th len socket =

338−#if NO_TIMEOUT_PROTECTION

343+ inner

339− bs <- liftIO $ Sock.recv socket len

340− go bs

341−#else

342− mbs <- liftIO $ timeout readTimeout $ Sock.recv socket len

343− case mbs of

344− Nothing -> throwError SocketTimeout

345− Just bs -> go bs

346−#endif

347 where

344 where

345+ inner (E.Continue k) = do

346+ bs <- liftIO $ Sock.recv socket len

347+ liftIO $ T.tickle th

348+ if S.null bs

349+ then E.throwError SocketTimeout

350+ else go k bs

351+ inner step = E.returnI step

348− go bs

352+ go k bs

349 | S.length bs == 0 = E.continue k

353 | S.length bs == 0 = E.continue k

350− | otherwise = k (E.Chunks [bs]) >>== enumSocket len socket

354+ | otherwise = k (E.Chunks [bs]) >>== enumSocket th len socket

351−enumSocket _ _ step = E.returnI step

352

355

353------ The functions below are not warp-specific and could be split out into a

356------ The functions below are not warp-specific and could be split out into a

354--separate package.

357--separate package.

@@ -360,7 +363,7 @@ takeUntilBlank :: Int

360 -> ([ByteString] -> [ByteString])

363 -> ([ByteString] -> [ByteString])

361 -> E.Iteratee S.ByteString IO [ByteString]

364 -> E.Iteratee S.ByteString IO [ByteString]

362takeUntilBlank count _

365takeUntilBlank count _

363− | count > maxHeaders = throwError TooManyHeaders

366+ | count > maxHeaders = E.throwError TooManyHeaders

364takeUntilBlank count front = do

367takeUntilBlank count front = do

365 l <- takeLineMax 0 id

368 l <- takeLineMax 0 id

366 if B.null l

369 if B.null l

@@ -373,7 +376,7 @@ takeLineMax :: Int

373takeLineMax len front = do

376takeLineMax len front = do

374 mbs <- EL.head

377 mbs <- EL.head

375 case mbs of

378 case mbs of

376− Nothing -> throwError IncompleteHeaders

379+ Nothing -> E.throwError IncompleteHeaders

377 Just bs -> do

380 Just bs -> do

378 let (x, y) = S.breakByte 10 bs

381 let (x, y) = S.breakByte 10 bs

379 x' = if S.length x > 0 && S.last x == 13

382 x' = if S.length x > 0 && S.last x == 13

@@ -382,10 +385,21 @@ takeLineMax len front = do

382 let len' = len + B.length x

385 let len' = len + B.length x

383 case () of

386 case () of

384 ()

387 ()

385− | len' > maxHeaderLength -> throwError OverLargeHeader

388+ | len' > maxHeaderLength -> E.throwError OverLargeHeader

386 | B.null y -> takeLineMax len' $ front . (:) x

389 | B.null y -> takeLineMax len' $ front . (:) x

387 | otherwise -> do

390 | otherwise -> do

388 E.yield () $ E.Chunks [B.drop 1 y]

391 E.yield () $ E.Chunks [B.drop 1 y]

389 return $ B.concat $ front [x']

392 return $ B.concat $ front [x']

390

393

391−throwError = liftIO . throwIO

394+iterSocket :: T.Handle

395+ -> Socket

396+ -> E.Iteratee B.ByteString IO ()

397+iterSocket th sock =

398+ E.continue step

399+ where

400+ step E.EOF = E.yield () E.EOF

401+ step (E.Chunks []) = E.continue step

402+ step (E.Chunks xs) = do

403+ liftIO $ Sock.sendMany sock xs

404+ liftIO $ T.tickle th

405+ E.continue step

Timeout.hsnew file+43

@@ -0,0 +1,43 @@

1+module Timeout

2+ ( Manager

3+ , Handle

4+ , initialize

5+ , register

6+ , tickle

7+ ) where

9+import qualified Data.IORef as I

10+import Control.Concurrent (forkIO, threadDelay)

11+import Control.Monad (forever)

12+

13+newtype Manager = Manager (I.IORef [Handle])

14+data Handle = Handle (IO ()) (I.IORef Bool)

15+

16+initialize :: Int -> IO Manager

17+initialize timeout = do

18+ ref <- I.newIORef []

19+ _ <- forkIO $ forever $ do

20+ threadDelay timeout

21+ ms <- I.atomicModifyIORef ref (\x -> ([], x))

22+ ms' <- go ms id

23+ I.atomicModifyIORef ref (\x -> (ms' x, ()))

24+ return $ Manager ref

25+ where

26+ go [] front = return front

27+ go (m@(Handle onTimeout iactive):rest) front = do

28+ active <- I.atomicModifyIORef iactive (\x -> (False, x))

29+ if active

30+ then go rest (front . (:) m)

31+ else do

32+ onTimeout

33+ go rest front

34+

35+register :: Manager -> IO () -> IO Handle

36+register (Manager ref) onTimeout = do

37+ iactive <- I.newIORef True

38+ let h = Handle onTimeout iactive

39+ I.atomicModifyIORef ref (\x -> (h : x, ()))

40+ return h

41+

42+tickle :: Handle -> IO ()

43+tickle (Handle _ iactive) = I.writeIORef iactive True

warp.cabal+2−1

@@ -1,5 +1,5 @@

1Name: warp

2−Version: 0.3.1

2+Version: 0.3.2

3Synopsis: A fast, light-weight web server for WAI applications.

4License: BSD3

5License-file: LICENSE

@@ -32,6 +32,7 @@ Library

32 else

33 build-depends: network >= 2.3 && < 2.4

34 Exposed-modules: Network.Wai.Handler.Warp

35+ Other-modules: Timeout

35 ghc-options: -Wall

36 ghc-options: -Wall

36 if !flag(timeout-protection)

37 if !flag(timeout-protection)

37 Cpp-options: -DNO_TIMEOUT_PROTECTION

38 Cpp-options: -DNO_TIMEOUT_PROTECTION