store 0.2.1.2 → 0.3
raw patch · 10 files changed
+641/−272 lines, 10 filesdep +asyncdep +freedep +networkdep −faildep ~store-core
Dependencies added: async, free, network, streaming-commons
Dependencies removed: fail
Dependency ranges changed: store-core
Files
- ChangeLog.md +14/−0
- README.md +0/−8
- src/Data/Store/Internal.hs +6/−5
- src/Data/Store/Streaming.hs +123/−99
- src/Data/Store/Streaming/Internal.hs +26/−0
- src/System/IO/ByteBuffer.hs +291/−83
- store.cabal +47/−36
- test/Data/Store/StreamingSpec.hs +102/−34
- test/Data/StoreSpec.hs +17/−6
- test/System/IO/ByteBufferSpec.hs +15/−1
ChangeLog.md view
@@ -1,5 +1,19 @@ # ChangeLog +## 0.3++* Uses store-core-0.3.*, which has support for alignment sensitive+ architectures.++* Adds support for streaming decode from file descriptor, not supported on+ windows. As part of this addition, the API for "Data.Store.Streaming" has+ changed.++## 0.2.1.2++* Fixes a bug that could could result in attempting to malloc a negative+ number of bytes when reading corrupted data.+ ## 0.2.1.1 * Fixes a bug that could result in segfaults when reading corrupted data.
README.md view
@@ -33,14 +33,6 @@ * TH generation of testcases. -## Architecture limitations--`Store` does not currently work at all on architectures which lack efficient-unaligned memory access (for example, older ARM processors). This is not a-fundamental limitation, but we do not currently require ARM or PowerPC support.-See [#37](https://github.com/fpco/store/issues/37) and-[#47](https://github.com/fpco/store/issues/47).- ## Blog posts * [Initial release announcement](https://www.fpcomplete.com/blog/2016/05/store-package)
src/Data/Store/Internal.hs view
@@ -273,9 +273,9 @@ -- | Skip n bytes forward. {-# INLINE skip #-} skip :: Int -> Peek ()-skip len = Peek $ \end ptr -> do+skip len = Peek $ \ps ptr -> do let ptr2 = ptr `plusPtr` len- remaining = end `minusPtr` ptr+ remaining = peekStateEndPtr ps `minusPtr` ptr when (len > remaining) $ -- Do not perform the check on the new pointer, since it could have overflowed tooManyBytes len remaining "skip" return (ptr2, ())@@ -284,12 +284,13 @@ -- advances the offset beyond the isolated region. {-# INLINE isolate #-} isolate :: Int -> Peek a -> Peek a-isolate len m = Peek $ \end ptr -> do- let ptr2 = ptr `plusPtr` len+isolate len m = Peek $ \ps ptr -> do+ let end = peekStateEndPtr ps+ ptr2 = ptr `plusPtr` len remaining = end `minusPtr` ptr when (len > remaining) $ -- Do not perform the check on the new pointer, since it could have overflowed tooManyBytes len remaining "isolate"- (ptr', x) <- runPeek m end ptr+ (ptr', x) <- runPeek m ps ptr when (ptr' > end) $ throwIO $ PeekException (ptr' `minusPtr` end) "Overshot end of isolated bytes" return (ptr2, x)
src/Data/Store/Streaming.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-| Module: Data.Store.Streaming Description: A thin streaming layer that uses 'Store' for serialisation.@@ -24,154 +26,176 @@ -- * Encoding 'Message's , encodeMessage -- * Decoding 'Message's- , PeekMessage (..)+ , PeekMessage+ , FillByteBuffer , peekMessage , decodeMessage+ , peekMessageBS+ , decodeMessageBS+#ifndef mingw32_HOST_OS+ , ReadMoreData(..)+ , peekMessageFd+ , decodeMessageFd+#endif -- * Conduits for encoding and decoding , conduitEncode , conduitDecode ) where -import Control.Exception (assert, throwIO)-import Control.Monad (liftM)+import Control.Exception (throwIO)+import Control.Monad (unless) import Control.Monad.IO.Class import Control.Monad.Trans.Resource (MonadResource) import Data.ByteString (ByteString)-import qualified Data.ByteString.Internal as BS import qualified Data.Conduit as C import qualified Data.Conduit.List as C import Data.Store import Data.Store.Impl (getSize)-import Data.Store.Core (Poke(..), tooManyBytes, decodeIOWithFromPtr)+import Data.Store.Core (decodeIOWithFromPtr, unsafeEncodeWith) import qualified Data.Text as T import Data.Word import Foreign.Ptr-import qualified Foreign.Storable as Storable import Prelude import System.IO.ByteBuffer (ByteBuffer) import qualified System.IO.ByteBuffer as BB+import Control.Monad.Trans.Free.Church (FT, iterTM, wrap)+import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)+import Control.Monad.Trans.Class (lift)+import System.Posix.Types (Fd(..))+import GHC.Conc (threadWaitRead)+import Data.Store.Streaming.Internal -- | If @a@ is an instance of 'Store', @Message a@ can be serialised -- and deserialised in a streaming fashion. newtype Message a = Message { fromMessage :: a } deriving (Eq, Show) --- | Type used to store the length of a 'Message'.-type SizeTag = Int---- | Some fixed arbitrary magic number that precedes every 'Message'.-messageMagic :: Word64-messageMagic = 18205256374652458875--magicLength :: Int-magicLength = Storable.sizeOf messageMagic--sizeTagLength :: Int-sizeTagLength = Storable.sizeOf (undefined :: SizeTag)--headerLength :: Int-headerLength = magicLength + sizeTagLength- -- | Encode a 'Message' to a 'ByteString'. encodeMessage :: Store a => Message a -> ByteString encodeMessage (Message x) =- let l = getSize x- totalLength = headerLength + l- in BS.unsafeCreate- totalLength- (\p -> do (offset, ()) <- runPoke (do poke messageMagic- poke l- poke x- ) p 0- assert (offset == totalLength) (return ()))+ unsafeEncodeWith pokeFunc totalLength+ where+ bodyLength = getSize x+ totalLength = headerLength + bodyLength+ pokeFunc = do+ poke messageMagic+ poke bodyLength+ poke x {-# INLINE encodeMessage #-} -- | The result of peeking at the next message can either be a--- successfully deserialised 'Message', or a request for more input.-data PeekMessage m a = Done (Message a)- | NeedMoreInput (ByteString -> m (PeekMessage m a))+-- successfully deserialised object, or a request for more input.+type PeekMessage i m a = FT ((->) i) m a --- | Decode a 'Message' of known size from a 'ByteBuffer'.-peekSized :: (MonadIO m, Store a) => ByteBuffer -> Int -> m (PeekMessage m a)-peekSized bb n =- BB.unsafeConsume bb n >>= \case- Right ptr -> liftM (Done . Message) $ decodeFromPtr ptr n- Left _ -> return $ NeedMoreInput (\ bs -> BB.copyByteString bb bs- >> peekSized bb n)+needMoreInput :: Monad m => PeekMessage i m i+needMoreInput = wrap return+{-# INLINE needMoreInput #-}++-- | Given some sort of input, fills the 'ByteBuffer' with it.+--+-- The 'Int' is how many bytes we'd like: this is useful when the filling+-- function is 'fillFromFd', where we can specify a max size.+type FillByteBuffer i m = ByteBuffer -> Int -> i -> m ()++-- | Decode a value, given a 'Ptr' and the number of bytes that make+-- up the encoded message.+decodeFromPtr :: (MonadIO m, Store a) => Ptr Word8 -> Int -> m a+decodeFromPtr ptr n = liftIO $ decodeIOWithFromPtr peek ptr n+{-# INLINE decodeFromPtr #-}++peekSized :: (MonadIO m, Store a) => FillByteBuffer i m -> ByteBuffer -> Int -> PeekMessage i m a+peekSized fill bb n = go+ where+ go = do+ mbPtr <- BB.unsafeConsume bb n+ case mbPtr of+ Left needed -> do+ inp <- needMoreInput+ lift (fill bb needed inp)+ go+ Right ptr -> decodeFromPtr ptr n {-# INLINE peekSized #-} -- | Decode a header (magic number and 'SizeTag') from a 'ByteBuffer'.-peekMessageHeader :: MonadIO m => ByteBuffer -> m (PeekMessage m SizeTag)-peekMessageHeader bb = do- available <- BB.availableBytes bb- if available < headerLength- then return $ NeedMoreInput (\bs -> BB.copyByteString bb bs- >> peekMessageHeader bb)- else peekSized bb magicLength >>= \case- Done (Message x) | x == messageMagic -> peekSized bb sizeTagLength- Done (Message x) -> liftIO . throwIO $ PeekException 0 . T.pack $ "Wrong message magic, " ++ show x- NeedMoreInput _ -> fail "Internal error in peekMessageHeader."+peekMessageHeader :: MonadIO m => FillByteBuffer i m -> ByteBuffer -> PeekMessage i m SizeTag+peekMessageHeader fill bb = go+ where+ go = do+ messageMagic' <- peekSized fill bb magicLength+ unless (messageMagic == messageMagic') $+ liftIO . throwIO $ PeekException 0 . T.pack $ "Wrong message magic, " ++ show messageMagic'+ peekSized fill bb sizeTagLength {-# INLINE peekMessageHeader #-} --- | Decode some 'Message' from a 'ByteBuffer', by first reading its--- header, and then the actual 'Message'.-peekMessage :: (MonadIO m, Store a) => ByteBuffer -> m (PeekMessage m a)-peekMessage bb =- peekMessageHeader bb >>= \case- (Done (Message n)) -> peekSized bb n- NeedMoreInput _ ->- return $ NeedMoreInput (\ bs -> BB.copyByteString bb bs- >> peekMessage bb)+-- | Decode some object from a 'ByteBuffer', by first reading its+-- header, and then the actual data.+peekMessage :: (MonadIO m, Store a) => FillByteBuffer i m -> ByteBuffer -> PeekMessage i m (Message a)+peekMessage fill bb =+ fmap Message (peekSized fill bb =<< peekMessageHeader fill bb) {-# INLINE peekMessage #-} -- | Decode a 'Message' from a 'ByteBuffer' and an action that can get--- additional 'ByteString's to refill the buffer when necessary.+-- additional inputs to refill the buffer when necessary. -- -- The only conditions under which this function will give 'Nothing', -- is when the 'ByteBuffer' contains zero bytes, and refilling yields -- 'Nothing'. If there is some data available, but not enough to -- decode the whole 'Message', a 'PeekException' will be thrown.-decodeMessage :: (MonadIO m, Store a)- => ByteBuffer -> m (Maybe ByteString) -> m (Maybe (Message a))-decodeMessage bb getBs =- decodeHeader bb getBs >>= \case- Nothing -> return Nothing- Just n -> decodeSized bb getBs n+decodeMessage :: (Store a, MonadIO m) => FillByteBuffer i m -> ByteBuffer -> m (Maybe i) -> m (Maybe (Message a))+decodeMessage fill bb getInp = do+ mbRes <- runMaybeT (iterTM (\consumeInp -> consumeInp =<< MaybeT getInp) (peekMessage fill bb))+ case mbRes of+ Just x -> return (Just x)+ Nothing -> do+ available <- BB.availableBytes bb+ unless (available == 0) $ liftIO $ throwIO $ PeekException available $ T.pack $+ "Data.Store.Streaming.decodeMessage: could not get enough bytes to decode message"+ return Nothing {-# INLINE decodeMessage #-} -decodeHeader :: MonadIO m- => ByteBuffer- -> m (Maybe ByteString)- -> m (Maybe SizeTag)-decodeHeader bb getBs =- peekMessageHeader bb >>= \case- (Done (Message n)) -> return (Just n)- (NeedMoreInput _) -> getBs >>= \case- Just bs -> BB.copyByteString bb bs >> decodeHeader bb getBs- Nothing -> BB.availableBytes bb >>= \case- 0 -> return Nothing- n -> liftIO $ tooManyBytes headerLength n "Data.Store.Message.SizeTag"-{-# INLINE decodeHeader #-}+-- | Decode some 'Message' from a 'ByteBuffer', by first reading its+-- header, and then the actual 'Message'.+peekMessageBS :: (MonadIO m, Store a) => ByteBuffer -> PeekMessage ByteString m (Message a)+peekMessageBS = peekMessage (\bb _ bs -> BB.copyByteString bb bs)+{-# INLINE peekMessageBS #-} -decodeSized :: (MonadIO m, Store a)- => ByteBuffer- -> m (Maybe ByteString)- -> Int- -> m (Maybe (Message a))-decodeSized bb getBs n =- peekSized bb n >>= \case- Done message -> return (Just message)- NeedMoreInput _ -> getBs >>= \case- Just bs -> BB.copyByteString bb bs >> decodeSized bb getBs n- Nothing -> BB.availableBytes bb >>= \ available ->- liftIO $ tooManyBytes n available "Data.Store.Message.Message"-{-# INLINE decodeSized #-}+decodeMessageBS :: (MonadIO m, Store a)+ => ByteBuffer -> m (Maybe ByteString) -> m (Maybe (Message a))+decodeMessageBS = decodeMessage (\bb _ bs -> BB.copyByteString bb bs)+{-# INLINE decodeMessageBS #-} --- | Decode a value, given a 'Ptr' and the number of bytes that make--- up the encoded message.-decodeFromPtr :: (MonadIO m, Store a) => Ptr Word8 -> Int -> m a-decodeFromPtr ptr n = liftIO $ decodeIOWithFromPtr peek ptr n-{-# INLINE decodeFromPtr #-}+#ifndef mingw32_HOST_OS +-- | We use this type as a more descriptive unit to signal that more input+-- should be read from the Fd.+--+-- This data-type is only available on POSIX systems (essentially, non-windows)+data ReadMoreData = ReadMoreData+ deriving (Eq, Show)++-- | Peeks a message from a _non blocking_ 'Fd'.+--+-- This function is only available on POSIX systems (essentially, non-windows)+peekMessageFd :: (MonadIO m, Store a) => ByteBuffer -> Fd -> PeekMessage ReadMoreData m (Message a)+peekMessageFd bb fd =+ peekMessage (\bb_ needed ReadMoreData -> do _ <- BB.fillFromFd bb_ fd needed; return ()) bb+{-# INLINE peekMessageFd #-}++-- | Decodes all the message using 'registerFd' to find out when a 'Socket' is+-- ready for reading.+--+-- This function is only available on POSIX systems (essentially, non-windows)+decodeMessageFd :: (MonadIO m, Store a) => ByteBuffer -> Fd -> m (Message a)+decodeMessageFd bb fd = do+ mbMsg <- decodeMessage+ (\bb_ needed ReadMoreData -> do _ <- BB.fillFromFd bb_ fd needed; return ()) bb+ (liftIO (threadWaitRead fd) >> return (Just ReadMoreData))+ case mbMsg of+ Just msg -> return msg+ Nothing -> liftIO (fail "decodeMessageFd: impossible: got Nothing")+{-# INLINE decodeMessageFd #-}++#endif+ -- | Conduit for encoding 'Message's to 'ByteString's. conduitEncode :: (Monad m, Store a) => C.Conduit (Message a) m ByteString conduitEncode = C.map encodeMessage@@ -191,7 +215,7 @@ go where go buffer = do- mmessage <- decodeMessage buffer C.await+ mmessage <- decodeMessageBS buffer C.await case mmessage of Nothing -> return () Just message -> C.yield message >> go buffer
+ src/Data/Store/Streaming/Internal.hs view
@@ -0,0 +1,26 @@+module Data.Store.Streaming.Internal+ ( messageMagic+ , magicLength+ , sizeTagLength+ , headerLength+ , SizeTag+ ) where++import Data.Word (Word64)+import qualified Foreign.Storable as Storable++-- | Type used to store the length of a 'Message'.+type SizeTag = Int++-- | Some fixed arbitrary magic number that precedes every 'Message'.+messageMagic :: Word64+messageMagic = 18205256374652458875++magicLength :: Int+magicLength = Storable.sizeOf messageMagic++sizeTagLength :: Int+sizeTagLength = Storable.sizeOf (undefined :: SizeTag)++headerLength :: Int+headerLength = sizeTagLength + magicLength
src/System/IO/ByteBuffer.hs view
@@ -1,6 +1,11 @@-{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-@ LIQUID "--no-termination" @-}+{-@ LIQUID "--short-names" @-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE CPP #-} {-| Module: System.IO.ByteBuffer Description: Provides an efficient buffering abstraction.@@ -26,25 +31,34 @@ , totalSize, isEmpty, availableBytes -- * Feeding new input , copyByteString+#ifndef mingw32_HOST_OS+ , fillFromFd+#endif -- * Consuming bytes from the buffer , consume, unsafeConsume+ -- * Exceptions+ , ByteBufferException (..) ) where import Control.Applicative-import Control.Exception.Lifted (bracket)-import Control.Monad (when)-import Control.Monad.IO.Class+import Control.Exception (SomeException, throwIO)+import Control.Exception.Lifted (Exception, bracket, catch)+import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Trans.Control (MonadBaseControl) import Data.ByteString (ByteString) import qualified Data.ByteString.Internal as BS import Data.IORef import Data.Maybe (fromMaybe)+import Data.Typeable (Typeable) import Data.Word import Foreign.ForeignPtr import qualified Foreign.Marshal.Alloc as Alloc-import Foreign.Marshal.Utils hiding (new, with)+import Foreign.Marshal.Utils (copyBytes, moveBytes) import GHC.Ptr import Prelude+import qualified Foreign.C.Error as CE+import Foreign.C.Types+import System.Posix.Types (Fd (..)) -- | A buffer into which bytes can be written. --@@ -61,6 +75,22 @@ -- data that have been copied to it, but not yet read. They are in -- the range from @ptr `plusPtr` consumedBytes@ to @ptr `plusPtr` -- containedBytes@.+--+-- The first two invariants are encoded in Liquid Haskell, and can+-- be statically checked.+--+-- If an Exception occurs during an operation that modifies a+-- 'ByteBuffer', the 'ByteBuffer' is invalidated and can no longer be+-- used. Trying to access the 'ByteBuffer' subsequently will result+-- in a 'ByteBufferException' being thrown.+{-@+data BBRef = BBRef+ { size :: {v: Int | v >= 0 }+ , contained :: { v: Int | v >= 0 && v <= size }+ , consumed :: { v: Int | v >= 0 && v <= contained }+ , ptr :: { v: Ptr Word8 | (plen v) = size }+ }+@-} data BBRef = BBRef { size :: {-# UNPACK #-} !Int@@ -74,10 +104,53 @@ -- the 'ByteBuffer' } -type ByteBuffer = IORef BBRef+-- | Exception that is thrown when an invalid 'ByteBuffer' is being used that is no longer valid.+--+-- A 'ByteBuffer' is considered to be invalid if+--+-- - it has explicitly been freed+-- - an Exception has occured during an operation that modified it+data ByteBufferException = ByteBufferException+ { _bbeLocation :: !String+ -- ^ function name that caused the exception+ , _bbeException :: !String+ -- ^ printed representation of the exception+ }+ deriving (Typeable, Eq)+instance Show ByteBufferException where+ show (ByteBufferException loc e) = concat+ ["ByteBufferException: ByteBuffer was invalidated because of Exception thrown in "+ , loc , ": ", e]+instance Exception ByteBufferException +type ByteBuffer = IORef (Either ByteBufferException BBRef)++-- | On any Exception, this will invalidate the ByteBuffer and re-throw the Exception.+--+-- Invalidating the 'ByteBuffer' includes freeing the underlying pointer.+bbHandler :: MonadIO m+ => String+ -- ^ location information: function from which the exception was thrown+ -> ByteBuffer+ -- ^ this 'ByteBuffer' will be invalidated when an Exception occurs+ -> (BBRef -> IO a)+ -> m a+bbHandler loc bb f = liftIO $ useBBRef f bb `catch` \(e :: SomeException) -> do+ readIORef bb >>= \case+ Right bbref -> do+ Alloc.free (ptr bbref)+ writeIORef bb (Left $ ByteBufferException loc (show e))+ Left _ -> return ()+ throwIO e+{-# INLINE bbHandler #-}++-- | Try to use the 'BBRef' of a 'ByteBuffer', or throw a 'ByteBufferException' if it's invalid.+useBBRef :: (BBRef -> IO a) -> ByteBuffer -> IO a+useBBRef f bb = readIORef bb >>= either throwIO f+{-# INLINE useBBRef #-}+ totalSize :: MonadIO m => ByteBuffer -> m Int-totalSize bb = liftIO $ size <$> readIORef bb+totalSize = liftIO . useBBRef (return . size) {-# INLINE totalSize #-} isEmpty :: MonadIO m => ByteBuffer -> m Bool@@ -86,20 +159,11 @@ -- | Number of available bytes in a 'ByteBuffer' (that is, bytes that -- have been copied to, but not yet read from the 'ByteBuffer'.+{-@ availableBytes :: MonadIO m => ByteBuffer -> m {v: Int | v >= 0} @-} availableBytes :: MonadIO m => ByteBuffer -> m Int-availableBytes bb = do- BBRef{..} <- liftIO $ readIORef bb- return $ contained - consumed+availableBytes = liftIO . useBBRef (\BBRef{..} -> return (contained - consumed)) {-# INLINE availableBytes #-} --- | The number of bytes that can be appended to the buffer, without--- resetting it.-freeCapacity :: MonadIO m => ByteBuffer -> m Int-freeCapacity bb = do- BBRef{..} <- liftIO $ readIORef bb- return $ size - contained-{-# INLINE freeCapacity #-}- -- | Allocates a new ByteBuffer with a given buffer size filling from -- the given FillBuffer. --@@ -113,18 +177,24 @@ -> m ByteBuffer -- ^ The byte buffer. new ml = liftIO $ do- let l = fromMaybe (4*1024*1024) ml+ let l = max 0 . fromMaybe (4*1024*1024) $ ml newPtr <- Alloc.mallocBytes l- newIORef BBRef { ptr = newPtr- , size = l- , contained = 0- , consumed = 0- }+ newIORef $ Right BBRef+ { ptr = newPtr+ , size = l+ , contained = 0+ , consumed = 0+ } {-# INLINE new #-} -- | Free a byte buffer. free :: MonadIO m => ByteBuffer -> m ()-free bb = liftIO $ readIORef bb >>= Alloc.free . ptr+free bb = liftIO $ readIORef bb >>= \case+ Right bbref -> do+ Alloc.free $ ptr bbref+ writeIORef bb $+ Left (ByteBufferException "free" "ByteBuffer has explicitly been freed and is no longer valid.")+ Left _ -> return () -- the ByteBuffer is either invalid or has already been freed. {-# INLINE free #-} -- | Perform some action with a bytebuffer, with automatic allocation@@ -142,70 +212,155 @@ action {-# INLINE with #-} --- | Reset a 'ByteBuffer', i.e. copy all the bytes that have not yet+-- | Reset a 'BBRef', i.e. copy all the bytes that have not yet -- been consumed to the front of the buffer.-reset :: ByteBuffer -> IO ()-reset bb = do- bbref@BBRef{..} <- readIORef bb- let available = contained - consumed- moveBytes ptr (ptr `plusPtr` consumed) available- writeIORef bb bbref { contained = available- , consumed = 0- }-{-# INLINE reset #-}+{-@ resetBBRef :: b:BBRef -> IO {v:BBRef | consumed v == 0 && contained v == contained b - consumed b && size v == size b} @-}+resetBBRef :: BBRef -> IO BBRef+resetBBRef bbref = do+ let available = contained bbref - consumed bbref+ moveBytes (ptr bbref) (ptr bbref `plusPtr` consumed bbref) available+ return BBRef { size = size bbref+ , contained = available+ , consumed = 0+ , ptr = ptr bbref+ }+{-# INLINE resetBBRef #-} -- | Make sure the buffer is at least @minSize@ bytes long. ----- In order to avoid havong to enlarge the buffer too often, we double--- its size until it is at least @minSize@ bytes long.-enlargeByteBuffer :: ByteBuffer- -> Int- -- ^ minSize- -> IO ()-enlargeByteBuffer bb minSize = do- bbref@BBRef{..} <- readIORef bb- when (size < minSize) $ do- let newSize = head . dropWhile (<minSize) $- iterate (ceiling . (*(1.5 :: Double)) . fromIntegral) (max 1 (size))+-- In order to avoid having to enlarge the buffer too often, we+-- multiply its size by a factor of 1.5 until it is at least @minSize@+-- bytes long.+{-@ enlargeBBRef :: b:BBRef -> i:Nat -> IO {v:BBRef | size v >= i && contained v == contained b && consumed v == consumed b} @-}+enlargeBBRef :: BBRef -> Int -> IO BBRef+enlargeBBRef bbref minSize = do+ let getNewSize s | s >= minSize = s+ getNewSize s = getNewSize $ (ceiling . (*(1.5 :: Double)) . fromIntegral) (max 1 s)+ newSize = getNewSize (size bbref) -- possible optimisation: since reallocation might copy the -- bytes anyway, we could discard the consumed bytes, -- basically 'reset'ting the buffer on the fly.- -- ptr' <- Alloc.mallocBytes newSize- ptr' <- Alloc.reallocBytes ptr newSize- writeIORef bb bbref { ptr = ptr'- , size = newSize- }-{-# INLINE enlargeByteBuffer #-}+ ptr' <- Alloc.reallocBytes (ptr bbref) newSize+ return BBRef { size = newSize+ , contained = contained bbref+ , consumed = consumed bbref+ , ptr = ptr'+ }+{-# INLINE enlargeBBRef #-} -- | Copy the contents of a 'ByteString' to a 'ByteBuffer'. -- -- If necessary, the 'ByteBuffer' is enlarged and/or already consumed -- bytes are dropped. copyByteString :: MonadIO m => ByteBuffer -> ByteString -> m ()-copyByteString bb bs@(BS.PS _ _ bsSize) = liftIO $ do- BBRef{..} <- readIORef bb- -- if the byteBuffer is too small, resize it.- available <- availableBytes bb -- bytes not yet consumed- when (size < bsSize + available) (enlargeByteBuffer bb (bsSize + available))- -- if it is currently too full, reset it- capacity <- freeCapacity bb- when (capacity < bsSize) (reset bb)- -- now we can safely copy.- unsafeCopyByteString bb bs+copyByteString bb bs =+ bbHandler "copyByteString" bb go+ where+ go bbref = do+ let (bsFptr, bsOffset, bsSize) = BS.toForeignPtr bs+ -- if the byteBuffer is too small, resize it.+ let available = contained bbref - consumed bbref -- bytes not yet consumed+ bbref' <- if size bbref < bsSize + available+ then enlargeBBRef bbref (bsSize + available)+ else return bbref+ -- if it is currently too full, reset it+ bbref'' <- if bsSize + contained bbref' > size bbref'+ then resetBBRef bbref'+ else return bbref'+ -- now we can safely copy.+ withForeignPtr bsFptr $ \ bsPtr ->+ copyBytes (ptr bbref'' `plusPtr` contained bbref'')+ (bsPtr `plusPtr` bsOffset)+ bsSize+ writeIORef bb $ Right BBRef {+ size = size bbref''+ , contained = contained bbref'' + bsSize+ , consumed = consumed bbref''+ , ptr = ptr bbref''} {-# INLINE copyByteString #-} --- | Copy the contents of a 'ByteString' to a 'ByteBuffer'. No bounds--- checks are performed.-unsafeCopyByteString :: ByteBuffer -> ByteString -> IO ()-unsafeCopyByteString bb (BS.PS bsFptr bsOffset bsSize) = do- bbref@BBRef{..} <- readIORef bb- withForeignPtr bsFptr $ \ bsPtr ->- copyBytes (ptr `plusPtr` contained)- (bsPtr `plusPtr` bsOffset)- bsSize- writeIORef bb bbref { contained = (contained + bsSize) }-{-# INLINE unsafeCopyByteString #-}+#ifndef mingw32_HOST_OS +-- | Will read at most n bytes from the given 'Fd', in a non-blocking+-- fashion. This function is intended to be used with non-blocking 'Socket's,+-- such the ones created by the @network@ package.+--+-- Returns how many bytes could be read non-blockingly.+fillFromFd :: MonadIO m => ByteBuffer -> Fd -> Int -> m Int+fillFromFd bb sock maxBytes = if maxBytes < 0+ then fail ("fillFromFd: negative argument (" ++ show maxBytes ++ ")")+ else bbHandler "fillFromFd" bb go+ where+ go bbref = do+ (bbref', readBytes) <- fillBBRefFromFd sock bbref maxBytes+ writeIORef bb $ Right bbref'+ return readBytes+{-# INLINE fillFromFd #-}++{-+Note: I'd like to use these two definitions:++{-@ measure _available @-}+_available :: BBRef -> Int+_available BBRef{..} = contained - consumed++{-@ measure _free @-}+_free :: BBRef -> Int+_free BBRef{..} = size - contained++but for some reason when I try to do so it does not work.+-}++{-@ fillBBRefFromFd ::+ Fd -> b0: BBRef+ -> maxBytes: Nat -> IO {v: (BBRef, Nat) | maxBytes >= snd v && contained (fst v) - consumed (fst v) == contained b0 - consumed b0 + snd v}+ @-}+fillBBRefFromFd :: Fd -> BBRef -> Int -> IO (BBRef, Int)+fillBBRefFromFd (Fd sock) bbref0 maxBytes = do+ bbref1 <- prepareSpace bbref0+ go 0 bbref1+ where+ -- We enlarge the buffer directly to be able to contain the maximum number+ -- of bytes since the common pattern for this function is to know how many+ -- bytes we need to read -- so we'll eventually fill the enlarged buffer.+ {-@ prepareSpace :: b: BBRef -> IO {v: BBRef | size v - contained v >= maxBytes && contained b - consumed b == contained v - consumed v} @-}+ prepareSpace :: BBRef -> IO BBRef+ prepareSpace bbref = do+ let space = size bbref - contained bbref+ if space < maxBytes+ then if consumed bbref > 0+ then prepareSpace =<< resetBBRef bbref+ else enlargeBBRef bbref (contained bbref + maxBytes)+ else return bbref++ {-@ go ::+ readBytes: {v: Nat | v <= maxBytes}+ -> b: {b: BBRef | size b - contained b >= maxBytes - readBytes}+ -> IO {v: (BBRef, Nat) | maxBytes >= snd v && snd v >= readBytes && size (fst v) - contained (fst v) >= maxBytes - snd v && contained (fst v) - consumed (fst v) == (contained b - consumed b) + (snd v - readBytes)}+ @-}+ go :: Int -> BBRef -> IO (BBRef, Int)+ go readBytes bbref@BBRef{..} = if readBytes >= maxBytes+ then return (bbref, readBytes)+ else do+ bytes <- fromIntegral <$> c_recv sock (castPtr (ptr `plusPtr` contained)) (fromIntegral (maxBytes - readBytes)) 0+ if bytes == -1+ then do+ err <- CE.getErrno+ if err == CE.eAGAIN || err == CE.eWOULDBLOCK+ then return (bbref, readBytes)+ else throwIO $ CE.errnoToIOError "ByteBuffer.fillBBRefFromFd: " err Nothing Nothing+ else do+ let bbref' = bbref{ contained = contained + bytes }+ go (readBytes + bytes) bbref'+{-# INLINE fillBBRefFromFd #-}++foreign import ccall unsafe "recv"+ -- c_recv returns -1 in the case of errors.+ {-@ assume c_recv :: CInt -> Ptr CChar -> size: {v: CSize | v >= 0} -> flags: CInt -> IO {read: CInt | read >= -1 && size >= read} @-}+ c_recv :: CInt -> Ptr CChar -> CSize -> CInt -> IO CInt++#endif+ -- | Try to get a pointer to @n@ bytes from the 'ByteBuffer'. -- -- Note that the pointer should be used before any other actions are@@ -213,6 +368,7 @@ -- buffer, so operations such as enlarging the buffer or feeding it -- new data will change the data the pointer points to. This is why -- this function is called unsafe.+{-@ unsafeConsume :: MonadIO m => ByteBuffer -> n:Nat -> m (Either Int ({v:Ptr Word8 | plen v >= n})) @-} unsafeConsume :: MonadIO m => ByteBuffer -> Int@@ -220,20 +376,23 @@ -> m (Either Int (Ptr Word8)) -- ^ Will be @Left missing@ when there are only @n-missing@ -- bytes left in the 'ByteBuffer'.-unsafeConsume bb n = liftIO $ do- available <- availableBytes bb- if available < n- then return $ Left (n - available)- else do- bbref@BBRef{..} <- readIORef bb- writeIORef bb bbref { consumed = (consumed + n) }- return $ Right (ptr `plusPtr` consumed)+unsafeConsume bb n =+ bbHandler "unsafeConsume" bb go+ where+ go bbref = do+ let available = contained bbref - consumed bbref+ if available < n+ then return $ Left (n - available)+ else do+ writeIORef bb $ Right bbref { consumed = consumed bbref + n }+ return $ Right (ptr bbref `plusPtr` consumed bbref) {-# INLINE unsafeConsume #-} -- | As `unsafeConsume`, but instead of returning a `Ptr` into the -- contents of the `ByteBuffer`, it returns a `ByteString` containing -- the next @n@ bytes in the buffer. This involves allocating a new -- 'ByteString' and copying the @n@ bytes to it.+{-@ consume :: MonadIO m => ByteBuffer -> Nat -> m (Either Int ByteString) @-} consume :: MonadIO m => ByteBuffer -> Int@@ -242,8 +401,57 @@ mPtr <- unsafeConsume bb n case mPtr of Right ptr -> do- bs <- liftIO . BS.create n $ \ bsPtr ->- copyBytes bsPtr ptr n+ bs <- liftIO $ createBS ptr n return (Right bs) Left missing -> return (Left missing) {-# INLINE consume #-}++{-@ createBS :: p:(Ptr Word8) -> {v:Nat | v <= plen p} -> IO ByteString @-}+createBS :: Ptr Word8 -> Int -> IO ByteString+createBS ptr n = do+ fp <- mallocForeignPtrBytes n+ withForeignPtr fp (\p -> copyBytes p ptr n)+ return (BS.PS fp 0 n)+{-# INLINE createBS #-}++-- below are liquid haskell qualifiers, and specifications for external functions.++{-@ qualif FPLenPLen(v:Ptr a, fp:ForeignPtr a): fplen fp = plen v @-}++{-@ Foreign.Marshal.Alloc.mallocBytes :: l:Nat -> IO (PtrN a l) @-}+{-@ Foreign.Marshal.Alloc.reallocBytes :: Ptr a -> l:Nat -> IO (PtrN a l) @-}+{-@ assume mallocForeignPtrBytes :: n:Nat -> IO (ForeignPtrN a n) @-}+{-@ type ForeignPtrN a N = {v:ForeignPtr a | fplen v = N} @-}+{-@ Foreign.Marshal.Utils.copyBytes :: p:Ptr a -> q:Ptr a -> {v:Nat | v <= plen p && v <= plen q} -> IO ()@-}+{-@ Foreign.Marshal.Utils.moveBytes :: p:Ptr a -> q:Ptr a -> {v:Nat | v <= plen p && v <= plen q} -> IO ()@-}+{-@ Foreign.Ptr.plusPtr :: p:Ptr a -> n:Nat -> {v:Ptr b | plen v == (plen p) - n} @-}++-- writing down the specification for ByteString is not as straightforward as it seems at first: the constructor+--+-- PS (ForeignPtr Word8) Int Int+--+-- has actually four arguments after unboxing (the ForeignPtr is an+-- Addr# and ForeignPtrContents), so restriciting the length of the+-- ForeignPtr directly in the specification of the datatype does not+-- work. Instead, I chose to write a specification for toForeignPtr.+-- It seems that the liquidhaskell parser has problems with variables+-- declared in a tuple, so I have to define the following measures to+-- get at the ForeignPtr, offset, and length.+--+-- This is a bit awkward, maybe there is an easier way.++_get1 :: (a,b,c) -> a+_get1 (x,_,_) = x+{-@ measure _get1 @-}+_get2 :: (a,b,c) -> b+_get2 (_,x,_) = x+{-@ measure _get2 @-}+_get3 :: (a,b,c) -> c+_get3 (_,_,x) = x+{-@ measure _get3 @-}++{-@ Data.ByteString.Internal.toForeignPtr :: ByteString ->+ {v:(ForeignPtr Word8, Int, Int) | _get2 v >= 0+ && _get2 v <= (fplen (_get1 v))+ && _get3 v >= 0+ && ((_get3 v) + (_get2 v)) <= (fplen (_get1 v))} @-}
store.cabal view
@@ -2,18 +2,18 @@ -- -- see: https://github.com/sol/hpack -name: store-version: 0.2.1.2-synopsis: Fast binary serialization-homepage: https://github.com/fpco/store#readme-bug-reports: https://github.com/fpco/store/issues-license: MIT-license-file: LICENSE-maintainer: Michael Sloan <sloan@fpcomplete.com>-copyright: 2016 FP Complete-category: Serialization, Data-build-type: Simple-cabal-version: >= 1.10+name: store+version: 0.3+synopsis: Fast binary serialization+category: Serialization, Data+homepage: https://github.com/fpco/store#readme+bug-reports: https://github.com/fpco/store/issues+maintainer: Michael Sloan <sloan@fpcomplete.com>+copyright: 2016 FP Complete+license: MIT+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10 extra-source-files: ChangeLog.md@@ -28,27 +28,16 @@ default: False flag small-bench- default: False manual: True+ default: False library hs-source-dirs: src- exposed-modules:- Data.Store- Data.Store.Internal- Data.Store.Streaming- Data.Store.TH- Data.Store.TH.Internal- Data.Store.TypeHash- Data.Store.TypeHash.Internal- Data.Store.Version- System.IO.ByteBuffer- other-modules:- Data.Store.Impl+ ghc-options: -Wall -fwarn-tabs -fwarn-incomplete-uni-patterns -fwarn-incomplete-record-updates -O2 build-depends: base >=4.7 && <5- , store-core >=0.2.0.2 && <0.3+ , store-core >=0.3 && <0.4 , th-utilities >=0.2 , primitive >=0.6 , th-reify-many >=0.1.6@@ -61,7 +50,6 @@ , cryptohash >=0.11.6 , deepseq >=1.3.0.2 , directory >= 1.2- , fail >=4.9.0.0 , filepath >= 1.3 , ghc-prim >=0.3.1.0 , hashable >=1.2.3.1@@ -86,10 +74,24 @@ , unordered-containers >=0.2.5.1 , vector >=0.10.12.3 , void >=0.5.11- if (!arch(I386) && !arch(X86_64) && !arch(IA64) && !impl(ghcjs))- buildable: False+ , free >=4.11+ , network >=2.6.0.2+ , streaming-commons >=0.1.10.0+ , async >=2.0.2+ exposed-modules:+ Data.Store+ Data.Store.Internal+ Data.Store.Streaming+ Data.Store.Streaming.Internal+ Data.Store.TH+ Data.Store.TH.Internal+ Data.Store.TypeHash+ Data.Store.TypeHash.Internal+ Data.Store.Version+ System.IO.ByteBuffer+ other-modules:+ Data.Store.Impl default-language: Haskell2010- ghc-options: -Wall -fwarn-tabs -fwarn-incomplete-uni-patterns -fwarn-incomplete-record-updates -O2 test-suite store-test type: exitcode-stdio-1.0@@ -99,7 +101,7 @@ ghc-options: -Wall -fwarn-tabs -fwarn-incomplete-uni-patterns -fwarn-incomplete-record-updates -O2 -threaded -rtsopts -with-rtsopts=-N build-depends: base >=4.7 && <5- , store-core >=0.2.0.2 && <0.3+ , store-core >=0.3 && <0.4 , th-utilities >=0.2 , primitive >=0.6 , th-reify-many >=0.1.6@@ -112,7 +114,6 @@ , cryptohash >=0.11.6 , deepseq >=1.3.0.2 , directory >= 1.2- , fail >=4.9.0.0 , filepath >= 1.3 , ghc-prim >=0.3.1.0 , hashable >=1.2.3.1@@ -137,6 +138,10 @@ , unordered-containers >=0.2.5.1 , vector >=0.10.12.3 , void >=0.5.11+ , free >=4.11+ , network >=2.6.0.2+ , streaming-commons >=0.1.10.0+ , async >=2.0.2 , store other-modules: Data.Store.StreamingSpec@@ -153,7 +158,7 @@ ghc-options: -Wall -fwarn-tabs -fwarn-incomplete-uni-patterns -fwarn-incomplete-record-updates -O2 -threaded -rtsopts -with-rtsopts=-N -with-rtsopts=-T -O2 build-depends: base >=4.7 && <5- , store-core >=0.2.0.2 && <0.3+ , store-core >=0.3 && <0.4 , th-utilities >=0.2 , primitive >=0.6 , th-reify-many >=0.1.6@@ -166,7 +171,6 @@ , cryptohash >=0.11.6 , deepseq >=1.3.0.2 , directory >= 1.2- , fail >=4.9.0.0 , filepath >= 1.3 , ghc-prim >=0.3.1.0 , hashable >=1.2.3.1@@ -191,6 +195,10 @@ , unordered-containers >=0.2.5.1 , vector >=0.10.12.3 , void >=0.5.11+ , free >=4.11+ , network >=2.6.0.2+ , streaming-commons >=0.1.10.0+ , async >=2.0.2 , store , weigh , criterion@@ -213,7 +221,7 @@ ghc-options: -Wall -fwarn-tabs -fwarn-incomplete-uni-patterns -fwarn-incomplete-record-updates -O2 -threaded -rtsopts -with-rtsopts=-N1 -with-rtsopts=-s -with-rtsopts=-qg build-depends: base >=4.7 && <5- , store-core >=0.2.0.2 && <0.3+ , store-core >=0.3 && <0.4 , th-utilities >=0.2 , primitive >=0.6 , th-reify-many >=0.1.6@@ -226,7 +234,6 @@ , cryptohash >=0.11.6 , deepseq >=1.3.0.2 , directory >= 1.2- , fail >=4.9.0.0 , filepath >= 1.3 , ghc-prim >=0.3.1.0 , hashable >=1.2.3.1@@ -251,6 +258,10 @@ , unordered-containers >=0.2.5.1 , vector >=0.10.12.3 , void >=0.5.11+ , free >=4.11+ , network >=2.6.0.2+ , streaming-commons >=0.1.10.0+ , async >=2.0.2 , criterion , store if flag(comparison-bench)
test/Data/Store/StreamingSpec.hs view
@@ -1,21 +1,34 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE CPP #-} module Data.Store.StreamingSpec where +import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (race, concurrently)+import Control.Concurrent.MVar import Control.Exception (try)-import Control.Monad (void)+import Control.Monad (void, (<=<), forM_, unless)+import Control.Monad.Trans.Free (runFreeT, FreeF(..))+import Control.Monad.Trans.Free.Church (fromFT) import Control.Monad.Trans.Resource import qualified Data.ByteString as BS-import qualified Data.ByteString.Internal as BS import Data.Conduit ((=$=), ($$)) import qualified Data.Conduit.List as C import Data.Int import Data.List (unfoldr) import Data.Monoid-import Data.Store.Core (Poke(..))+import Data.Store.Core (unsafeEncodeWith) import Data.Store.Internal import Data.Store.Streaming+import Data.Store.Streaming.Internal+import Data.Streaming.Network (runTCPServer, runTCPClient, clientSettingsTCP, serverSettingsTCP, setAfterBind)+import Data.Streaming.Network.Internal (AppData(..))+import Data.Void (absurd, Void)+import Network.Socket (Socket(..), socketPort)+import Network.Socket.ByteString (send) import qualified System.IO.ByteBuffer as BB+import System.Posix.Types (Fd(..)) import Test.Hspec import Test.Hspec.SmallCheck import Test.SmallCheck@@ -28,14 +41,20 @@ it "Throws an Exception on incomplete messages." conduitIncomplete it "Throws an Exception on excess input." $ property conduitExcess describe "peekMessage" $ do- it "demands more input when needed." $ property (askMore 17)- it "demands more input on incomplete message magic." $ property (askMore 1)- it "demands more input on incomplete SizeTag." $ property (askMore 9)- it "successfully decodes valid input." $ property canPeek+ describe "ByteString" $ do+ it "demands more input when needed." $ property (askMoreBS (headerLength + 1))+ it "demands more input on incomplete message magic." $ property (askMoreBS 1)+ it "demands more input on incomplete SizeTag." $ property (askMoreBS (headerLength - 1))+ it "successfully decodes valid input." $ property canPeekBS describe "decodeMessage" $ do- it "Throws an Exception on incomplete messages." decodeIncomplete- it "Throws an Exception on messages that are shorter than indicated." decodeTooShort- it "Throws an Exception on messages that are longer than indicated." decodeTooLong+ describe "ByteString" $ do+ it "Throws an Exception on incomplete messages." decodeIncomplete+ it "Throws an Exception on messages that are shorter than indicated." decodeTooShort+ it "Throws an Exception on messages that are longer than indicated." decodeTooLong+#ifndef mingw32_HOST_OS+ describe "Socket" $ do+ it "Decodes data trickling through a socket." $ property decodeTricklingMessageFd+#endif roundtrip :: [Int] -> Property IO roundtrip xs = monadic $ do@@ -91,32 +110,85 @@ -- peekResult, expecting it to require more input. Then, feeds the -- second part and checks if the decoded result is the original -- message.-askMore :: Int -> Integer -> Property IO-askMore n x = monadic $ BB.with (Just 10) $ \ bb -> do+askMoreBS :: Int -> Integer -> Property IO+askMoreBS n x = monadic $ BB.with (Just 10) $ \ bb -> do let bs = encodeMessage (Message x) (start, end) = BS.splitAt n $ bs BB.copyByteString bb start- peekResult <- peekMessage bb :: IO (PeekMessage IO Integer)+ peekResult <- runFreeT (fromFT (peekMessageBS bb)) case peekResult of- NeedMoreInput cont ->- cont end >>= \case- Done (Message x') -> return $ x' == x- _ -> return False- _ -> return False+ Free cont ->+ runFreeT (cont end) >>= \case+ Pure (Message x') -> return $ x' == x+ Free _ -> return False+ Pure _ -> return False -canPeek :: Integer -> Property IO-canPeek x = monadic $ BB.with (Just 10) $ \ bb -> do+canPeekBS :: Integer -> Property IO+canPeekBS x = monadic $ BB.with (Just 10) $ \ bb -> do let bs = encodeMessage (Message x) BB.copyByteString bb bs- peekResult <- peekMessage bb :: IO (PeekMessage IO Integer)+ peekResult <- runFreeT (fromFT (peekMessageBS bb)) case peekResult of- NeedMoreInput _ -> return False- Done (Message x') -> return $ x' == x+ Free _ -> return False+ Pure (Message x') -> return $ x' == x +#ifndef mingw32_HOST_OS++socketFd :: Socket -> Fd+socketFd (MkSocket fd _ _ _ _) = Fd fd++withServer :: (Socket -> Socket -> IO a) -> IO a+withServer cont = do+ sock1Var :: MVar Socket <- newEmptyMVar+ sock2Var :: MVar Socket <- newEmptyMVar+ portVar :: MVar Int <- newEmptyMVar+ doneVar :: MVar Void <- newEmptyMVar+ let adSocket ad = case appRawSocket' ad of+ Nothing -> error "withServer.adSocket: no raw socket in AppData"+ Just sock -> sock+ let ss = setAfterBind+ (putMVar portVar . fromIntegral <=< socketPort)+ (serverSettingsTCP 0 "127.0.0.1")+ x <- fmap (either (either absurd absurd) id) $ race+ (race+ (runTCPServer ss $ \ad -> do+ putMVar sock1Var (adSocket ad)+ void (readMVar doneVar))+ (do port <- takeMVar portVar+ runTCPClient (clientSettingsTCP port "127.0.0.1") $ \ad -> do+ putMVar sock2Var (adSocket ad)+ readMVar doneVar))+ (do sock1 <- takeMVar sock1Var+ sock2 <- takeMVar sock2Var+ cont sock1 sock2)+ putMVar doneVar (error "withServer: impossible: read from doneVar")+ return x++decodeTricklingMessageFd :: Integer -> Property IO+decodeTricklingMessageFd x = monadic $ do+ let bs = encodeMessage (Message x)+ BB.with Nothing $ \bb ->+ withServer $ \sock1 sock2 -> do+ let generateChunks :: [Int] -> BS.ByteString -> [BS.ByteString]+ generateChunks xs0 bs_ = case xs0 of+ [] -> generateChunks [1,3,10] bs_+ x : xs -> if BS.null bs_+ then []+ else BS.take x bs_ : generateChunks xs (BS.drop x bs_)+ let chunks = generateChunks [] bs+ ((), Message x') <- concurrently+ (forM_ chunks $ \chunk -> do+ void (send sock1 chunk)+ threadDelay (10 * 1000))+ (decodeMessageFd bb (socketFd sock2))+ return (x == x')++#endif+ decodeIncomplete :: IO () decodeIncomplete = BB.with (Just 0) $ \ bb -> do BB.copyByteString bb (BS.take 1 incompleteInput)- (decodeMessage bb (return Nothing) :: IO (Maybe (Message Integer)))+ (decodeMessageBS bb (return Nothing) :: IO (Maybe (Message Integer))) `shouldThrow` \PeekException{} -> True incompleteInput :: BS.ByteString@@ -127,27 +199,23 @@ decodeTooLong :: IO () decodeTooLong = BB.with Nothing $ \bb -> do BB.copyByteString bb (encodeMessageTooLong . Message $ (1 :: Int))- (decodeMessage bb (return Nothing) :: IO (Maybe (Message Int)))+ (decodeMessageBS bb (return Nothing) :: IO (Maybe (Message Int))) `shouldThrow` \PeekException{} -> True decodeTooShort :: IO () decodeTooShort = BB.with Nothing $ \bb -> do BB.copyByteString bb (encodeMessageTooShort . Message $ (1 :: Int))- (decodeMessage bb (return Nothing) :: IO (Maybe (Message Int)))- `shouldThrow` (== PeekException 8 "Attempted to read too many bytes for Data.Store.Message.SizeTag. Needed 16, but only 8 remain.")+ (decodeMessageBS bb (return Nothing) :: IO (Maybe (Message Int)))+ `shouldThrow` \PeekException{} -> True encodeMessageTooLong :: Store a => Message a -> BS.ByteString encodeMessageTooLong (Message x) = let l = 8 + getSize x totalLength = 8 + l- in BS.unsafeCreate- totalLength- (\p -> void $ runPoke (poke l >> poke x >> poke (0::Int64)) p 0)+ in unsafeEncodeWith (poke l >> poke x >> poke (0::Int64)) totalLength encodeMessageTooShort :: Store a => Message a -> BS.ByteString-encodeMessageTooShort (Message x) =+encodeMessageTooShort (Message _x) = let l = 0 totalLength = 8 + l- in BS.unsafeCreate- totalLength- (\p -> void $ runPoke (poke l >> poke x) p 0)+ in unsafeEncodeWith (poke l) totalLength
test/Data/StoreSpec.hs view
@@ -82,23 +82,34 @@ -- Serial instances for (Num a, Bounded a) types. Only really -- appropriate for the use here. -$(do let ns = [ ''CWchar, ''CUid, ''CUShort, ''CULong, ''CULLong, ''CIntMax+$(do let ns = [ ''CWchar, ''CUShort, ''CULong, ''CULLong, ''CIntMax , ''CUIntMax, ''CPtrdiff, ''CSChar, ''CShort, ''CUInt, ''CLLong- , ''CLong, ''CInt, ''CChar, ''CTcflag, ''CSsize, ''CRLim, ''CPid- , ''COff, ''CNlink, ''CMode, ''CIno, ''CGid, ''CDev+ , ''CLong, ''CInt, ''CChar, ''CSsize, ''CPid+ , ''COff, ''CMode, ''CIno, ''CDev , ''Word8, ''Word16, ''Word32, ''Word64, ''Word , ''Int8, ''Int16, ''Int32, ''Int64- ]+ ] +++#ifdef mingw32_HOST_OS+ []+#else+ [ ''CUid, ''CTcflag, ''CRLim, ''CNlink, ''CGid ]+#endif f n = [d| instance Monad m => Serial m $(conT n) where series = generate (\_ -> addMinAndMaxBounds [0, 1]) |] concat <$> mapM f ns) + -- Serial instances for (Num a) types. Only really appropriate for the -- use here. $(do let ns = [ ''CUSeconds, ''CClock, ''CTime, ''CUChar, ''CSize, ''CSigAtomic- , ''CSUSeconds, ''CFloat, ''CDouble, ''CSpeed, ''CCc- ]+ , ''CSUSeconds, ''CFloat, ''CDouble+ ] +++#ifdef mingw32_HOST_OS+ []+#else+ [ ''CSpeed, ''CCc ]+#endif f n = [d| instance Monad m => Serial m $(conT n) where series = generate (\_ -> [0, 1]) |] concat <$> mapM f ns)
test/System/IO/ByteBufferSpec.hs view
@@ -1,10 +1,17 @@+{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-} module System.IO.ByteBufferSpec where +import Control.Exception import qualified Data.ByteString as BS+import Data.Typeable (Typeable) import qualified System.IO.ByteBuffer as BB import Test.Hspec +data MyException = MyException+ deriving (Eq, Show, Typeable)+instance Exception MyException+ spec :: Spec spec = describe "ByteBuffer" $ do it "can grow to store a value and return it." $ BB.with (Just 0) $ \ bb -> do@@ -35,6 +42,13 @@ bbSize <- BB.totalSize bb bbSize `shouldBe` BS.length bs1 bbIsEmpty bb-+ it "should raise a ByteBufferException when used after freed" $ BB.with Nothing $ \bb -> do+ BB.free bb+ BB.totalSize bb `shouldThrow` \(BB.ByteBufferException loc e) ->+ loc == "free" && e == "ByteBuffer has explicitly been freed and is no longer valid."+ it "should raise a ByteBufferException after a failed operation" $ BB.with Nothing $ \bb -> do+ BB.copyByteString bb (throw MyException) `shouldThrow` (\MyException -> True)+ BB.consume bb 10 `shouldThrow` \(BB.ByteBufferException loc e) ->+ loc == "copyByteString" && e == show MyException bbIsEmpty :: BB.ByteBuffer -> Expectation bbIsEmpty bb = BB.isEmpty bb >>= (`shouldBe` True)