recv 0.1.0 → 0.1.1
raw patch · 8 files changed
+155/−175 lines, 8 files
Files
- ChangeLog.md +6/−0
- Network/Socket/BufferPool.hs +19/−16
- Network/Socket/BufferPool/Buffer.hs +14/−11
- Network/Socket/BufferPool/Recv.hs +55/−69
- Network/Socket/BufferPool/Types.hs +9/−8
- Network/Socket/BufferPool/Windows.hs +0/−25
- recv.cabal +49/−43
- test/BufferPoolSpec.hs +3/−3
+ ChangeLog.md view
@@ -0,0 +1,6 @@+# ChangeLog for "recv"++## v0.1.1++* Fixing the bug that the last chunk is skipped when the size is+ insufficient [1031](https://github.com/yesodweb/wai/pull/1031)
Network/Socket/BufferPool.hs view
@@ -21,22 +21,25 @@ -- When the buffer gets small -- and usefless, a new large buffer is allocated. module Network.Socket.BufferPool (- -- * Recv- Recv- , receive- , BufferPool- , newBufferPool- , withBufferPool- -- * RecvN- , RecvN- , makeRecvN- -- * Types- , Buffer- , BufSize- -- * Utilities- , mallocBS- , copy- ) where+ -- * Recv+ Recv,+ receive,+ BufferPool,+ newBufferPool,+ withBufferPool,++ -- * RecvN+ RecvN,+ makeRecvN,++ -- * Types+ Buffer,+ BufSize,++ -- * Utilities+ mallocBS,+ copy,+) where import Network.Socket.BufferPool.Buffer import Network.Socket.BufferPool.Recv
Network/Socket/BufferPool/Buffer.hs view
@@ -1,16 +1,17 @@ module Network.Socket.BufferPool.Buffer (- newBufferPool- , withBufferPool- , mallocBS- , copy- ) where+ newBufferPool,+ withBufferPool,+ mallocBS,+ copy,+) where import qualified Data.ByteString as BS-import Data.ByteString.Internal (ByteString(..), memcpy)-import Data.ByteString.Unsafe (unsafeTake, unsafeDrop)+import Data.ByteString.Internal (ByteString (..))+import Data.ByteString.Unsafe (unsafeDrop, unsafeTake) import Data.IORef (newIORef, readIORef, writeIORef) import Foreign.ForeignPtr-import Foreign.Marshal.Alloc (mallocBytes, finalizerFree)+import Foreign.Marshal.Alloc (finalizerFree, mallocBytes)+import Foreign.Marshal.Utils (copyBytes) import Foreign.Ptr (castPtr, plusPtr) import Network.Socket.BufferPool.Types@@ -35,8 +36,10 @@ withBufferPool :: BufferPool -> (Buffer -> BufSize -> IO Int) -> IO ByteString withBufferPool (BufferPool l h ref) f = do buf0 <- readIORef ref- buf <- if BS.length buf0 >= l then return buf0- else mallocBS h+ buf <-+ if BS.length buf0 >= l+ then return buf0+ else mallocBS h consumed <- withForeignBuffer buf f writeIORef ref $ unsafeDrop consumed buf return $ unsafeTake consumed buf@@ -59,6 +62,6 @@ -- 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)+ copyBytes ptr (p `plusPtr` o) (fromIntegral l) return $ ptr `plusPtr` l {-# INLINE copy #-}
Network/Socket/BufferPool/Recv.hs view
@@ -1,65 +1,24 @@-{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings #-}-{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-} module Network.Socket.BufferPool.Recv (- receive- , makeRecvN- ) where+ receive,+ makeRecvN,+) where import qualified Data.ByteString as BS-import Data.ByteString.Internal (ByteString(..), unsafeCreate)+import Data.ByteString.Internal (ByteString (..), unsafeCreate) import Data.IORef-import Foreign.C.Error (eAGAIN, getErrno, throwErrno)-import Foreign.C.Types-import Foreign.Ptr (Ptr, castPtr)-import GHC.Conc (threadWaitRead)-import Network.Socket (Socket, withFdSocket)-import System.Posix.Types (Fd(..))--#ifdef mingw32_HOST_OS-import GHC.IO.FD (FD(..), readRawBufferPtr)-import Network.Socket.BufferPool.Windows-#endif+import Network.Socket (Socket, recvBuf) -import Network.Socket.BufferPool.Types import Network.Socket.BufferPool.Buffer+import Network.Socket.BufferPool.Types ---------------------------------------------------------------- -- | The receiving function with a buffer pool. -- The buffer pool is automatically managed. receive :: Socket -> BufferPool -> Recv-receive sock pool = withBufferPool pool $ \ptr size -> do-#if MIN_VERSION_network(3,1,0)- withFdSocket sock $ \fd -> do-#elif MIN_VERSION_network(3,0,0)- fd <- fdSocket sock-#else- let fd = fdSocket sock-#endif- let size' = fromIntegral size- fromIntegral <$> tryReceive fd ptr size'--------------------------------------------------------------------tryReceive :: CInt -> Buffer -> CSize -> IO CInt-tryReceive sock ptr size = go- where- go = do-#ifdef mingw32_HOST_OS- bytes <- windowsThreadBlockHack $ fromIntegral <$> readRawBufferPtr "tryReceive" (FD sock 1) (castPtr ptr) 0 size-#else- bytes <- c_recv sock (castPtr ptr) size 0-#endif- if bytes == -1 then do- errno <- getErrno- if errno == eAGAIN then do- threadWaitRead (Fd sock)- go- else- throwErrno "tryReceive"- else- return bytes+receive sock pool = withBufferPool pool $ \ptr size -> recvBuf sock ptr size ---------------------------------------------------------------- @@ -71,6 +30,14 @@ -- When N is less than equal to 4096, the buffer pool is used. -- Otherwise, a new buffer is allocated. -- In this case, the global lock is taken.+--+-- >>> :seti -XOverloadedStrings+-- >>> tryRecvN "a" 3 =<< _iorefRecv ["bcd"]+-- ("abc","d")+-- >>> tryRecvN "a" 3 =<< _iorefRecv ["bc"]+-- ("abc","")+-- >>> tryRecvN "a" 3 =<< _iorefRecv ["b"]+-- ("ab","") makeRecvN :: ByteString -> Recv -> IO RecvN makeRecvN bs0 recv = do ref <- newIORef bs0@@ -89,34 +56,53 @@ tryRecvN :: ByteString -> Int -> IO ByteString -> IO (ByteString, ByteString) tryRecvN init0 siz0 recv- | siz0 <= len0 = return $ BS.splitAt siz0 init0- | otherwise = go (init0:) (siz0 - len0)+ | siz0 <= len0 = return $ BS.splitAt siz0 init0+ | otherwise = go (init0 :) (siz0 - len0) where len0 = BS.length init0 go build left = do bs <- recv let len = BS.length bs- if len == 0 then- return ("", "")- else if len >= left then do- let (consume, leftover) = BS.splitAt left bs- ret = concatN siz0 $ build [consume]- return (ret, leftover)- else do- let build' = build . (bs :)- left' = left - len- go build' left'+ if len == 0+ then do+ let cs = concatN (siz0 - left) $ build []+ return (cs, "")+ else+ if len >= left+ then do+ let (consume, leftover) = BS.splitAt left bs+ ret = concatN siz0 $ build [consume]+ return (ret, leftover)+ else do+ let build' = build . (bs :)+ left' = left - len+ go build' left' concatN :: Int -> [ByteString] -> ByteString-concatN total bss0 = unsafeCreate total $ \ptr -> goCopy bss0 ptr+-- Just because it's logical+concatN _ [] = ""+-- To avoid a copy if there's only one ByteString+concatN _ [bs] = bs+concatN total bss0 =+ unsafeCreate total $ \ptr -> goCopy bss0 ptr where- goCopy [] _ = return ()- goCopy (bs:bss) ptr = do+ goCopy [] _ = return ()+ goCopy (bs : bss) ptr = do ptr' <- copy ptr bs goCopy bss ptr' -#ifndef mingw32_HOST_OS--- fixme: the type of the return value-foreign import ccall unsafe "recv"- c_recv :: CInt -> Ptr CChar -> CSize -> CInt -> IO CInt-#endif+-- | doctest only. Elements in the argument must not be empty.+_iorefRecv :: [ByteString] -> IO (IO ByteString)+_iorefRecv ini = do+ ref <- newIORef ini+ return $ recv ref+ where+ recv ref = do+ xxs <- readIORef ref+ case xxs of+ [] -> do+ writeIORef ref $ error "closed"+ return ""+ x : xs -> do+ writeIORef ref xs+ return x
Network/Socket/BufferPool/Types.hs view
@@ -12,14 +12,15 @@ type BufSize = Int -- | Type for read buffer pool.-data BufferPool = BufferPool {- minBufSize :: Int -- ^ If the buffer is larger than or equal to this size,- -- the buffer is used.- -- Otherwise, a new buffer is allocated.- -- The thrown buffer is eventually freed.- , maxBufSize :: Int- , poolBuffer :: IORef ByteString- }+data BufferPool = BufferPool+ { minBufSize :: Int+ -- ^ If the buffer is larger than or equal to this size,+ -- the buffer is used.+ -- Otherwise, a new buffer is allocated.+ -- The thrown buffer is eventually freed.+ , maxBufSize :: Int+ , poolBuffer :: IORef ByteString+ } -- | Type for the receiving function with a buffer pool. type Recv = IO ByteString
− Network/Socket/BufferPool/Windows.hs
@@ -1,25 +0,0 @@-{-# LANGUAGE CPP #-}-module Network.Socket.BufferPool.Windows- ( windowsThreadBlockHack- ) where--#ifdef mingw32_HOST_OS-import Control.Concurrent.MVar-import Control.Concurrent-import qualified Control.Exception-import Control.Monad---- | Allow main socket listening thread to be interrupted on Windows platform-windowsThreadBlockHack :: IO a -> IO a-windowsThreadBlockHack act = do- var <- newEmptyMVar :: IO (MVar (Either Control.Exception.SomeException a))- -- Catch and rethrow even async exceptions, so don't bother with UnliftIO- void . forkIO $ Control.Exception.try act >>= putMVar var- res <- takeMVar var- case res of- Left e -> Control.Exception.throwIO e- Right r -> return r-#else-windowsThreadBlockHack :: IO a -> IO a-windowsThreadBlockHack = id-#endif
recv.cabal view
@@ -1,45 +1,51 @@-Name: recv-Version: 0.1.0-Synopsis: Efficient network recv-License: BSD3-License-file: LICENSE-Author: Kazu Yamamoto-Maintainer: kazu@iij.ad.jp-Homepage: http://github.com/yesodweb/wai-Category: Network-Build-Type: Simple-Cabal-Version: >= 1.10-Stability: Stable-description: Network recv based on buffer pools+cabal-version: >=1.10+name: recv+version: 0.1.1+license: BSD3+license-file: LICENSE+maintainer: kazu@iij.ad.jp+author: Kazu Yamamoto+stability: Stable+homepage: http://github.com/yesodweb/wai+synopsis: Efficient network recv+description: Network recv based on buffer pools+category: Network+build-type: Simple+extra-source-files: ChangeLog.md -Library- Build-Depends: base >= 4.12 && < 5- , bytestring >= 0.9.1.4- , network >= 3.1.0- Exposed-modules: Network.Socket.BufferPool- Other-modules: Network.Socket.BufferPool.Buffer- Network.Socket.BufferPool.Recv- Network.Socket.BufferPool.Types- Network.Socket.BufferPool.Windows- if impl(ghc >= 8)- Default-Extensions: Strict StrictData- Ghc-Options: -Wall- Default-Language: Haskell2010+library+ exposed-modules: Network.Socket.BufferPool+ other-modules:+ Network.Socket.BufferPool.Buffer+ Network.Socket.BufferPool.Recv+ Network.Socket.BufferPool.Types -Test-Suite spec- Main-Is: Spec.hs- Other-modules: BufferPoolSpec- Network.Socket.BufferPool- Network.Socket.BufferPool.Buffer- Network.Socket.BufferPool.Recv- Network.Socket.BufferPool.Types- Network.Socket.BufferPool.Windows- Hs-Source-Dirs: test, .- Type: exitcode-stdio-1.0- Build-Depends: base >= 4.12 && < 5- , bytestring >= 0.9.1.4- , network >= 3.1.0- , hspec- Ghc-Options: -Wall- Default-Language: Haskell2010- Build-Tool-Depends: hspec-discover:hspec-discover+ default-language: Haskell2010+ ghc-options: -Wall+ build-depends:+ base >=4.12 && <5,+ bytestring >=0.9.1.4,+ network >=3.1.0++ if impl(ghc >=8)+ default-extensions: Strict StrictData++test-suite spec+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ build-tool-depends: hspec-discover:hspec-discover+ hs-source-dirs: test .+ other-modules:+ BufferPoolSpec+ Network.Socket.BufferPool+ Network.Socket.BufferPool.Buffer+ Network.Socket.BufferPool.Recv+ Network.Socket.BufferPool.Types++ default-language: Haskell2010+ ghc-options: -Wall+ build-depends:+ base >=4.12 && <5,+ bytestring >=0.9.1.4,+ network >=3.1.0,+ hspec
test/BufferPoolSpec.hs view
@@ -1,20 +1,20 @@ module BufferPoolSpec where import qualified Data.ByteString as B-import qualified Data.ByteString.Internal as B (ByteString(PS))+import qualified Data.ByteString.Internal as B (ByteString (PS)) import Foreign.ForeignPtr (withForeignPtr) import Foreign.Marshal.Utils (copyBytes) import Foreign.Ptr (plusPtr) import Network.Socket.BufferPool-import Test.Hspec (Spec, hspec, shouldBe, describe, it)+import Test.Hspec (Spec, describe, hspec, it, shouldBe) main :: IO () main = hspec spec -- Two ByteStrings each big enough to fill a buffer (16K). wantData, otherData :: B.ByteString-wantData = B.replicate 16384 0xac+wantData = B.replicate 16384 0xac otherData = B.replicate 16384 0x77 spec :: Spec