concurrent-buffer (empty) → 0.1
raw patch · 8 files changed
+460/−0 lines, 8 filesdep +basedep +base-preludedep +bugsetup-changed
Dependencies added: base, base-prelude, bug, bytestring, concurrent-buffer, criterion, quickcheck-instances, rerebase, tasty, tasty-hunit, tasty-quickcheck
Files
- LICENSE +22/−0
- Setup.hs +2/−0
- benchmarks/Main.hs +28/−0
- concurrent-buffer.cabal +101/−0
- library/ConcurrentBuffer.hs +216/−0
- library/ConcurrentBuffer/Prelude.hs +22/−0
- library/ConcurrentBuffer/PtrIO.hs +11/−0
- tests/Main.hs +58/−0
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2017, Nikita Volkov++Permission is hereby granted, free of charge, to any person+obtaining a copy of this software and associated documentation+files (the "Software"), to deal in the Software without+restriction, including without limitation the rights to use,+copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following+conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmarks/Main.hs view
@@ -0,0 +1,28 @@+module Main where++import Prelude+import Bug+import Criterion.Main+import qualified ConcurrentBuffer as A+import qualified Data.ByteString as D+++main =+ defaultMain =<< sequence+ [+ benchPush "1" (2^8) 1+ ,+ benchPush "10" (2^8) 10+ ,+ benchPush "100" (2^8) 100+ ]++benchOnBuffer name size io =+ do+ buffer <- A.new size+ return (bench name (whnfIO (io buffer)))++benchPush name bufferSize factor =+ benchOnBuffer name bufferSize $ + let !bytes = fromString (replicate 10000 'a')+ in \buffer -> replicateM factor $ A.pushBytes buffer bytes
+ concurrent-buffer.cabal view
@@ -0,0 +1,101 @@+name:+ concurrent-buffer+version:+ 0.1+category:+ Data+synopsis:+ Concurrent expanding buffer+homepage:+ https://github.com/nikita-volkov/concurrent-buffer +bug-reports:+ https://github.com/nikita-volkov/concurrent-buffer/issues +author:+ Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer:+ Nikita Volkov <nikita.y.volkov@mail.ru>+copyright:+ (c) 2017, Nikita Volkov+license:+ MIT+license-file:+ LICENSE+build-type:+ Simple+cabal-version:+ >=1.10++source-repository head+ type:+ git+ location:+ git://github.com/nikita-volkov/concurrent-buffer.git++library+ hs-source-dirs:+ library+ default-extensions:+ Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, PatternSynonyms, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples+ default-language:+ Haskell2010+ exposed-modules:+ ConcurrentBuffer+ other-modules:+ ConcurrentBuffer.Prelude+ ConcurrentBuffer.PtrIO+ build-depends:+ -- data:+ bytestring >= 0.10 && < 0.11,+ -- errors:+ bug >= 1 && < 2,+ -- general:+ base-prelude >= 1 && < 2,+ base >= 4.7 && < 5++test-suite tests+ type:+ exitcode-stdio-1.0+ hs-source-dirs:+ tests+ main-is:+ Main.hs+ ghc-options:+ -O2+ -threaded+ "-with-rtsopts=-N"+ default-extensions:+ Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples+ default-language:+ Haskell2010+ build-depends:+ --+ concurrent-buffer,+ -- testing:+ tasty == 0.11.*,+ tasty-quickcheck == 0.8.*,+ tasty-hunit == 0.9.*,+ quickcheck-instances >= 0.3.11 && < 0.4,+ --+ bug == 1.*,+ rerebase == 1.*++benchmark benchmarks+ type:+ exitcode-stdio-1.0+ hs-source-dirs:+ benchmarks+ main-is:+ Main.hs+ ghc-options:+ -O2+ -threaded+ -rtsopts+ default-extensions:+ Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveTraversable, DeriveGeneric, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples+ default-language:+ Haskell2010+ build-depends:+ concurrent-buffer,+ criterion == 1.1.*,+ bug == 1.*,+ rerebase == 1.*
+ library/ConcurrentBuffer.hs view
@@ -0,0 +1,216 @@+module ConcurrentBuffer+(+ Buffer,+ new,+ push,+ pushBytes,+ pushStorable,+ pull,+ pullBytes,+ pullStorable,+ getSpace,+ getBytes,+)+where++import ConcurrentBuffer.Prelude hiding (State, Buffer)+import qualified ConcurrentBuffer.PtrIO as A+import qualified Data.ByteString.Internal as C+++data Buffer =+ {-|+ * Buffer pointer+ * Start offset+ * End offset+ * Capacity+ -}+ Buffer+ {-# UNPACK #-} !(TVar (ForeignPtr Word8))+ {-# UNPACK #-} !(TVar Int)+ {-# UNPACK #-} !(TVar Int)+ {-# UNPACK #-} !(TVar Int)+ {-# UNPACK #-} !(TVar Bool)+ {-# UNPACK #-} !(TVar Bool)+ {-# UNPACK #-} !(TVar Bool)+++{-|+Create a new buffer of the specified initial capacity.+-}+new :: Int -> IO Buffer+new capacity =+ do+ fptr <- mallocForeignPtrBytes capacity+ atomically $ do+ fptrVar <- newTVar fptr+ startVar <- newTVar 0+ endVar <- newTVar 0+ capVar <- newTVar capacity+ notPullingVar <- newTVar True+ notPushingVar <- newTVar True+ notAligningVar <- newTVar True+ return (Buffer fptrVar startVar endVar capVar notPullingVar notPushingVar notAligningVar)++{-|+Prepares the buffer to be filled with at maximum the specified amount of bytes,+then uses the pointer-action to populate it.+It is your responsibility to ensure that the action does not exceed the space limit.++The pointer-action returns the amount of bytes it actually writes to the buffer.+That amount then is used to move the buffer's cursor accordingly.+It can also produce some @result@, which will then be emitted by @push@.++It also aligns or grows the buffer if required.+-}+push :: Buffer -> Int -> (Ptr Word8 -> IO (Int, result)) -> IO result+push (Buffer fptrVar startVar endVar capVar notPullingVar notPushingVar notAligningVar) space ptrIO =+ join $ atomically $ do+ notPushing <- readTVar notPushingVar+ guard notPushing+ writeTVar notPushingVar False+ fptr <- readTVar fptrVar+ start <- readTVar startVar+ end <- readTVar endVar+ capacity <- readTVar capVar+ let+ !remainingSpace = capacity - end+ !capacityDelta = space - remainingSpace+ !occupiedSpace = end - start+ if capacityDelta <= 0 -- Doesn't need more space?+ then+ return $ do+ (!pushedSpace, output) <- withForeignPtr fptr $ \ptr -> ptrIO (plusPtr ptr end)+ atomically $ do+ writeTVar endVar $! end + pushedSpace+ writeTVar notPushingVar True+ return output+ else if capacityDelta > start -- Needs growing?+ then+ -- Grow+ return $ do+ let !newCapacity = occupiedSpace + space+ newFPtr <- mallocForeignPtrBytes newCapacity+ (!pushedSpace, output) <- withForeignPtr newFPtr $ \newPtr -> do+ withForeignPtr fptr $ \ptr -> A.memcpy newPtr (plusPtr ptr start) (fromIntegral occupiedSpace)+ ptrIO (plusPtr newPtr occupiedSpace)+ let !newEnd = occupiedSpace + pushedSpace+ atomically $ do+ divergedStart <- readTVar startVar+ let !newStart = divergedStart - start+ writeTVar fptrVar newFPtr+ writeTVar startVar newStart+ writeTVar endVar newEnd+ writeTVar capVar newCapacity+ writeTVar notPushingVar True+ return output+ else+ -- Align+ do+ notPulling <- readTVar notPullingVar+ guard notPulling+ writeTVar notAligningVar False+ return $ do+ (!pushedSpace, output) <- withForeignPtr fptr $ \ptr -> do+ A.memmove ptr (plusPtr ptr start) (fromIntegral occupiedSpace)+ atomically $ do+ writeTVar startVar 0+ writeTVar endVar occupiedSpace+ writeTVar notAligningVar True+ ptrIO (plusPtr ptr occupiedSpace)+ atomically $ do+ writeTVar endVar $! occupiedSpace + pushedSpace+ writeTVar notPushingVar True+ return output++{-|+Pulls the specified amount of bytes from the buffer using the provided pointer-action,+freeing the buffer from the pulled bytes afterwards.++In case the buffer does not contain enough bytes yet, it will block waiting.+-}+pull :: Buffer -> Int -> (Ptr Word8 -> IO result) -> IO result+pull (Buffer fptrVar startVar endVar capVar notPullingVar notPushingVar notAligningVar) amount ptrIO =+ join $ atomically $ do+ notPulling <- readTVar notPullingVar+ guard notPulling+ fptr <- readTVar fptrVar+ start <- readTVar startVar+ end <- readTVar endVar+ guard (amount <= end - start)+ notAligning <- readTVar notAligningVar+ guard notAligning+ writeTVar notPullingVar False+ return $ do+ pulled <- withForeignPtr fptr $ \ptr -> ptrIO (plusPtr ptr start)+ atomically $ do+ start <- readTVar startVar+ writeTVar startVar $! start + amount+ writeTVar notPullingVar True+ return pulled++{-|+Push a byte array into the buffer.+-}+{-# INLINE pushBytes #-}+pushBytes :: Buffer -> ByteString -> IO ()+pushBytes buffer (C.PS bytesFPtr offset length) =+ push buffer length $ \ptr ->+ withForeignPtr bytesFPtr $ \bytesPtr ->+ C.memcpy ptr (plusPtr bytesPtr offset) length $> (length, ())++{-|+Pulls the specified amount of bytes.+-}+{-# INLINE pullBytes #-}+pullBytes :: Buffer -> Int -> IO ByteString+pullBytes buffer amount =+ pull buffer amount (\ptr -> C.create amount (\destPtr -> C.memcpy destPtr ptr amount))++{-|+Push a storable value into the buffer.+-}+{-# INLINE pushStorable #-}+pushStorable :: (Storable storable) => Buffer -> storable -> IO ()+pushStorable buffer storable =+ push buffer amount (\ptr -> poke (castPtr ptr) storable $> (amount, ()))+ where+ amount = sizeOf storable++{-|+Pulls a storable value.+-}+{-# INLINE pullStorable #-}+pullStorable :: (Storable storable) => Buffer -> IO storable+pullStorable buffer =+ result+ where+ result =+ pull buffer amount (\ptr -> peek (castPtr ptr))+ where+ amount =+ sizeOf ((undefined :: IO a -> a) result)++{-|+Get how much space is occupied by the buffer's data.+-}+{-# INLINE getSpace #-}+getSpace :: Buffer -> IO Int+getSpace (Buffer fptrVar startVar endVar capVar notPullingVar notPushingVar notAligningVar) =+ atomically $ do+ end <- readTVar endVar+ start <- readTVar startVar+ return $! end - start++{-|+Create a bytestring representation without modifying the buffer.+-}+{-# INLINE getBytes #-}+getBytes :: Buffer -> IO ByteString+getBytes (Buffer fptrVar startVar endVar capVar notPullingVar notPushingVar notAligningVar) =+ join $ atomically $ do+ fptr <- readTVar fptrVar+ end <- readTVar endVar+ start <- readTVar startVar+ let size = end - start+ return $ withForeignPtr fptr $ \ptr -> C.create size $ \destPtr -> C.memcpy destPtr ptr size
+ library/ConcurrentBuffer/Prelude.hs view
@@ -0,0 +1,22 @@+module ConcurrentBuffer.Prelude+(+ module Exports,+)+where+++-- base+-------------------------+import Foreign.C as Exports++-- base-prelude+-------------------------+import BasePrelude as Exports hiding (assert, left, right, isLeft, isRight, (<>), First(..), Last(..), ProtocolError, traceEvent, traceEventIO, traceMarker, traceMarkerIO)++-- bytestring+-------------------------+import Data.ByteString as Exports (ByteString)++-- bug+-------------------------+import Bug as Exports
+ library/ConcurrentBuffer/PtrIO.hs view
@@ -0,0 +1,11 @@+module ConcurrentBuffer.PtrIO+where++import ConcurrentBuffer.Prelude+++foreign import ccall unsafe "memmove"+ memmove :: Ptr a {-^ Destination -} -> Ptr a {-^ Source -} -> CSize {-^ Count -} -> IO (Ptr a) {-^ Destination -}++foreign import ccall unsafe "memcpy"+ memcpy :: Ptr a {-^ Destination -} -> Ptr a {-^ Source -} -> CSize {-^ Count -} -> IO (Ptr a) {-^ Destination -}
+ tests/Main.hs view
@@ -0,0 +1,58 @@+module Main where++import Prelude+import Bug+import Test.Tasty+import Test.Tasty.Runners+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck+import Test.QuickCheck.Instances+import qualified ConcurrentBuffer as A+import qualified Data.ByteString as D+import qualified Data.ByteString.Internal as B+++main =+ defaultMain $+ testGroup "All tests"+ [+ testCase "Interleaving" $ do+ let (input1, input2, input3) = (1, 2, 3) :: (Word64, Word64, Word64)+ buffer <- A.new 2+ A.pushStorable buffer input1+ A.pushStorable buffer input2+ output1 <- A.pullStorable buffer+ A.pushStorable buffer input3+ output2 <- A.pullStorable buffer+ output3 <- A.pullStorable buffer+ assertEqual (show output1) input1 output1+ assertEqual (show output2) input2 output2+ assertEqual (show output3) input3 output3+ ,+ testProperty "Numbers" $ \(inputs :: [Word64]) ->+ unsafePerformIO $ do+ buffer <- A.new 2+ forM_ inputs $ \input -> A.pushStorable buffer input+ outputs <- (traverse (const (A.pullStorable buffer)) inputs)+ return (inputs === outputs)+ ,+ testProperty "Concatenation of pushed bytestrings equals the binary representation" $ \(inputs :: [ByteString]) ->+ unsafePerformIO $ do+ buffer <- A.new (2 ^ 8)+ forM_ inputs $ \input -> A.pushBytes buffer input+ let concattedInputs = mconcat inputs+ output <- A.pullBytes buffer (D.length concattedInputs)+ return (concattedInputs === output)+ ,+ testProperty "Concurrent push and pull equality" $ \(inputs :: Vector ByteString) ->+ unsafePerformIO $ do+ buffer <- A.new (2 ^ 8)+ let !amounts = D.length <$!> inputs+ forkIO $ forM_ inputs $ \input -> A.pushBytes buffer input+ outputs <- traverse (A.pullBytes buffer) amounts+ return (inputs == outputs)+ ]++inspect :: A.Buffer -> IO ()+inspect buffer =+ print . D.unpack =<< A.getBytes buffer