d7658fd · typeable/wai

git.iko.soy/typeable/wai

Receive buffering performance optimization, no memcpy and less malloc

commitd7658fdb0780bc7a075048d76cdfeddea42de784 authorMartin Bailey <martin@steamsharp.com> dateWed 17 Sep 2014, 15:44 (12 years ago) parent9de9805

4 files changed+68−36

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/Buffer.hs+48−2 warp/Network/Wai/Handler/Warp/Recv.hs+12−30 warp/Network/Wai/Handler/Warp/Run.hs+5−3 warp/Network/Wai/Handler/Warp/Types.hs+3−1

warp/Network/Wai/Handler/Warp/Buffer.hs+48−2

@@ -1,10 +1,56 @@

11module Network.Wai.Handler.Warp.Buffer where

3+import Control.Monad (when)

34import qualified Data.Streaming.ByteString.Builder.Buffer as B (Buffer (..))

45import Data.Word (Word8)

5−import Foreign.ForeignPtr (newForeignPtr_)

6+import Data.IORef (IORef, newIORef, readIORef, writeIORef)

7+import Data.ByteString (ByteString, empty, length)

8+import Data.ByteString.Internal (ByteString(PS), create)

9+import Data.ByteString.Unsafe (unsafeTake, unsafeDrop)

10+import Foreign.ForeignPtr (newForeignPtr_, withForeignPtr)

611import Foreign.Marshal.Alloc (mallocBytes, free)

7−import Foreign.Ptr (Ptr, plusPtr)

12+import Foreign.Ptr (Ptr, castPtr, plusPtr)

13+

14+largeBufferSize :: Int

15+largeBufferSize = 16384

16+minBufferSize :: Int

17+minBufferSize = 2048

18+

19+type BufferPool = IORef ByteString

20+

21+newBufferPool :: IO BufferPool

22+newBufferPool = newIORef Data.ByteString.empty

23+

24+allocBuffer :: Int -> IO ByteString

25+allocBuffer size = create size initializer

26+ where initializer = \_ -> return ()

27+{-# INLINE allocBuffer #-}

28+

29+usefulBuffer :: ByteString -> Bool

30+usefulBuffer buffer = Data.ByteString.length buffer >= minBufferSize

31+{-# INLINE usefulBuffer #-}

32+

33+getBuffer :: BufferPool -> IO ByteString

34+getBuffer pool = do

35+ buffer <- readIORef pool

36+ if usefulBuffer buffer then return buffer else allocBuffer largeBufferSize

37+{-# INLINE getBuffer #-}

38+

39+putBuffer :: BufferPool -> ByteString -> IO ()

40+putBuffer pool buffer = when (usefulBuffer buffer) $ writeIORef pool buffer

41+{-# INLINE putBuffer #-}

42+

43+withForeignBuffer :: ByteString -> ((Ptr Word8, Int) -> IO Int) -> IO Int

44+withForeignBuffer (PS ps s l) f = withForeignPtr ps $ \p -> f (castPtr p `plusPtr` s, l)

45+{-# INLINE withForeignBuffer #-}

46+

47+withBufferPool :: BufferPool -> ((Ptr Word8, Int) -> IO Int) -> IO ByteString

48+withBufferPool pool f = do

49+ buffer <- getBuffer pool

50+ consumed <- withForeignBuffer buffer f

51+ putBuffer pool $! unsafeDrop consumed buffer

52+ return $! unsafeTake consumed buffer

53+{-# INLINE withBufferPool #-}

854

955type Buffer = Ptr Word8

1056type BufSize = Int

warp/Network/Wai/Handler/Warp/Recv.hs+12−30

@@ -8,18 +8,15 @@ module Network.Wai.Handler.Warp.Recv (

88#if __GLASGOW_HASKELL__ < 709

99import Control.Applicative ((<$>))

1010#endif

11−import Control.Monad (void)

12−import qualified Data.ByteString as BS (empty)

13−import Data.ByteString.Internal (ByteString(..), mallocByteString)

11+import Data.ByteString (ByteString)

1412import Data.Word (Word8)

1513import Foreign.C.Error (eAGAIN, getErrno, throwErrno)

1614import Foreign.C.Types

17−import Foreign.ForeignPtr (withForeignPtr)

1815import Foreign.Ptr (Ptr, castPtr)

1916import GHC.Conc (threadWaitRead)

2017import Network.Socket (Socket, fdSocket)

21−import System.Posix.Types (Fd(..))

2218import Network.Wai.Handler.Warp.Buffer

19+import System.Posix.Types (Fd(..))

2320

2421#ifdef mingw32_HOST_OS

2522import GHC.IO.FD (FD(..), readRawBufferPtr)

@@ -28,37 +25,25 @@ import Network.Wai.Handler.Warp.Windows

2825

2926----------------------------------------------------------------

3027

31−receive :: Socket -> Buffer -> Int -> IO ByteString

32−receive sock buf size = do

33− bytes <- fromIntegral <$> receiveloop sock' buf' size'

34− if bytes == 0 then

35− return BS.empty

36− else do

37− fptr <- mallocByteString bytes

38− void $ withForeignPtr fptr $ \ptr ->

39− c_memcpy ptr buf (fromIntegral bytes)

40− return $! PS fptr 0 bytes

41− where

42− sock' = fdSocket sock

43− buf' = castPtr buf

44− size' = fromIntegral size

28+receive :: Socket -> BufferPool -> IO ByteString

29+receive sock pool = withBufferPool pool $ \ (ptr, size) -> do

30+ let sock' = fdSocket sock

31+ size' = fromIntegral size

32+ received <- fromIntegral <$> receiveloop sock' ptr size'

33+ return received

4534

46−#ifdef mingw32_HOST_OS

4735receiveloop :: CInt -> Ptr Word8 -> CSize -> IO CInt

48−#else

49−receiveloop :: CInt -> Ptr CChar -> CSize -> IO CInt

50−#endif

51−receiveloop sock buf size = do

36+receiveloop sock ptr size = do

5237#ifdef mingw32_HOST_OS

53− bytes <- windowsThreadBlockHack $ fmap fromIntegral $ readRawBufferPtr "recv" (FD sock 1) buf 0 size

38+ bytes <- windowsThreadBlockHack $ fmap fromIntegral $ readRawBufferPtr "recv" (FD sock 1) ptr 0 size

5439#else

55− bytes <- c_recv sock buf size 0

40+ bytes <- c_recv sock (castPtr ptr) size 0

5641#endif

5742 if bytes == -1 then do

5843 errno <- getErrno

5944 if errno == eAGAIN then do

6045 threadWaitRead (Fd sock)

61− receiveloop sock buf size

46+ receiveloop sock ptr size

6247 else

6348 throwErrno "receiveloop"

6449 else

@@ -67,6 +52,3 @@ receiveloop sock buf size = do

6752-- fixme: the type of the return value

6853foreign import ccall unsafe "recv"

6954 c_recv :: CInt -> Ptr CChar -> CSize -> CInt -> IO CInt

70−

71−foreign import ccall unsafe "string.h memcpy"

72− c_memcpy :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)

warp/Network/Wai/Handler/Warp/Run.hs+5−3

@@ -48,14 +48,16 @@ import Network.Socket (fdSocket)

4848-- | Default action value for 'Connection'.

4949socketConnection :: Socket -> IO Connection

5050socketConnection s = do

51− readBuf <- allocateBuffer bufferSize

51+ bufferPool <- newBufferPool

52+ readBuf <- allocateBuffer 0 -- kept for backward compatibility

5253 writeBuf <- allocateBuffer bufferSize

5354 return Connection {

5455 connSendMany = Sock.sendMany s

5556 , connSendAll = Sock.sendAll s

5657 , connSendFile = defaultSendFile s

57− , connClose = sClose s >> freeBuffer readBuf >> freeBuffer writeBuf

58− , connRecv = receive s readBuf bufferSize

58+ , connClose = sClose s >> freeBuffer writeBuf

59+ , connRecv = receive s bufferPool

60+ , connBufferPool = bufferPool

5961 , connReadBuffer = readBuf

6062 , connWriteBuffer = writeBuf

6163 , connBufferSize = bufferSize

warp/Network/Wai/Handler/Warp/Types.hs+3−1

@@ -12,7 +12,7 @@ import Data.Typeable (Typeable)

1212import Data.Word (Word16)

1313import Network.HTTP.Types.Header

1414import Network.Socket (Socket)

15−import Network.Wai.Handler.Warp.Buffer (Buffer,BufSize)

15+import Network.Wai.Handler.Warp.Buffer (Buffer,BufSize,BufferPool)

1616import qualified Network.Wai.Handler.Warp.Date as D

1717import qualified Network.Wai.Handler.Warp.FdCache as F

1818import qualified Network.Wai.Handler.Warp.Timeout as T

@@ -79,7 +79,9 @@ data Connection = Connection

7979 , connSendFile :: FilePath -> Integer -> Integer -> IO () -> [ByteString] -> IO () -- ^ filepath, offset, length, hook action, HTTP headers

8080 , connClose :: IO ()

8181 , connRecv :: IO ByteString

82+ , connBufferPool :: BufferPool

8283 , connReadBuffer :: Buffer

84+ -- ^ NOTE: This is kept just for backwards compatibility, and is no longer used at all.

8385 , connWriteBuffer :: Buffer

8486 , connBufferSize :: BufSize

8587 , connSendFileOverride :: ConnSendFileOverride