3375

pending completion

Build # 3375

Build Type

Pull #751

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/composewell/streamly/commit/<a class=hub.com/composewell/streamly/commit/b7943666ff7029def6e596702bdba242a0a6f90d">b7943666f<a href="https://github.com/composewell/streamly/commit/b7943666ff7029def6e596702bdba242a0a6f90d">">Merge </a><a class="double-link" href="https://github.com/composewell/streamly/commit/<a class="double-link" href="https://github.com/composewell/streamly/commit/63c83f9921b7ff2230594e5ec1a719f0b784792c">63c83f992</a>">63c83f992</a><a href="https://github.com/composewell/streamly/commit/b7943666ff7029def6e596702bdba242a0a6f90d"> into ebbaef79f">ebbaef79f</a>

Pull Request Pull Request #751: Refactor splitOnSeq

Run Details

162 of 162 new or added lines in 3 files covered. (100.0%)

3807 of 9607 relevant lines covered (39.63%)

0.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.56

/src/Streamly/Memory/Ring.hs

-- |
-- Module      : Streamly.Memory.Ring
-- Copyright   : (c) 2019 Composewell Technologies
-- License     : BSD3
-- Maintainer  : streamly@composewell.com
-- Stability   : experimental
-- Portability : GHC
--

module Streamly.Memory.Ring
    ( Ring(..)

    -- * Construction
    , new
    , advance
    , moveBy

    -- * Modification
    , unsafeInsert

    -- * Folds
    , unsafeFoldRing
    , unsafeFoldRingM
    , unsafeFoldRingFullM

    -- * Fast Byte Comparisons
    , unsafeEqArray
    , unsafeEqArrayN
    ) where

import Control.Exception (assert)
import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
import Foreign.Ptr (plusPtr, minusPtr, castPtr)
import Foreign.Storable (Storable(..))
import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
import GHC.Ptr (Ptr(..))
import Prelude hiding (length, concat)

import Control.Monad.IO.Class (MonadIO(..))

import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as A

-- | A ring buffer is a mutable array of fixed size. Initially the array is
-- empty, with ringStart pointing at the start of allocated memory. We call the
-- next location to be written in the ring as ringHead. Initially ringHead ==
-- ringStart. When the first item is added, ringHead points to ringStart +
-- sizeof item. When the buffer becomes full ringHead would wrap around to
-- ringStart. When the buffer is full, ringHead always points at the oldest
-- item in the ring and the newest item added always overwrites the oldest
-- item.
--
-- When using it we should keep in mind that a ringBuffer is a mutable data
-- structure. We should not leak out references to it for immutable use.
--
data Ring a = Ring
    { ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address
    , ringBound :: {-# UNPACK #-} !(Ptr a)        -- first address beyond allocated memory
    }

-- | Create a new ringbuffer and return the ring buffer and the ringHead.
-- Returns the ring and the ringHead, the ringHead is same as ringStart.
{-# INLINE new #-}
new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
new count = do
    let size = count * sizeOf (undefined :: a)
    fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))
    let p = unsafeForeignPtrToPtr fptr
    return (Ring
        { ringStart = fptr
        , ringBound = p `plusPtr` size
        }, p)

-- | Advance the ringHead by 1 item, wrap around if we hit the end of the
-- array.
{-# INLINE advance #-}
advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
advance Ring{..} ringHead =
    let ptr = ringHead `plusPtr` sizeOf (undefined :: a)
    in if ptr <  ringBound
       then ptr
       else unsafeForeignPtrToPtr ringStart

-- | Move the ringHead by n items. The direction depends on the sign on whether
-- n is positive or negative. Wrap around if we hit the beginning or end of the
-- array.
{-# INLINE moveBy #-}
moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead

    where

    elemSize = sizeOf (undefined :: a)
    ringStartPtr = unsafeForeignPtrToPtr ringStart
    lenInBytes = ringBound `minusPtr` ringStartPtr
    offInBytes = ringHead `minusPtr` ringStartPtr
    len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize
    off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize
    advanceFromHead = (off + by `mod` len) * elemSize

-- | Insert an item at the head of the ring, when the ring is full this
-- replaces the oldest item in the ring with the new item. This is unsafe
-- beause ringHead supplied is not verified to be within the Ring. Also,
-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
{-# INLINE unsafeInsert #-}
unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
unsafeInsert rb ringHead newVal = do
    poke ringHead newVal
    -- touchForeignPtr (ringStart rb)
    return $ advance rb ringHead

-- XXX remove all usage of A.unsafeInlineIO
--
-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of
-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
-- be in range.
{-# INLINE unsafeEqArrayN #-}
unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
unsafeEqArrayN Ring{..} rh A.Array{..} n =
    let !res = A.unsafeInlineIO $ do
            let rs = unsafeForeignPtrToPtr ringStart
                as = unsafeForeignPtrToPtr aStart
            assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())
            let len = ringBound `minusPtr` rh
            r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)
            r2 <- if n > len
                then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
                              (min (rh `minusPtr` rs) (n - len))
                else return True
            -- XXX enable these, check perf impact
            -- touchForeignPtr ringStart
            -- touchForeignPtr aStart
            return (r1 && r2)
    in res

-- | Byte compare the entire length of ringBuffer with the given array,
-- starting at the supplied ringHead pointer.  Returns true if the Array and
-- the ringBuffer have identical contents.
--
-- This is unsafe because the ringHead Ptr is not checked to be in range. The
-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
-- ARE NOT CHECKED.
{-# INLINE unsafeEqArray #-}
unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
unsafeEqArray Ring{..} rh A.Array{..} =
    let !res = A.unsafeInlineIO $ do
            let rs = unsafeForeignPtrToPtr ringStart
            let as = unsafeForeignPtrToPtr aStart
            assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)
                   (return ())
            let len = ringBound `minusPtr` rh
            r1 <- A.memcmp (castPtr rh) (castPtr as) len
            r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))
                           (rh `minusPtr` rs)
            -- XXX enable these, check perf impact
            -- touchForeignPtr ringStart
            -- touchForeignPtr aStart
            return (r1 && r2)
    in res

-- XXX use MonadIO
--
-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
-- step function. This is useful to fold the items in the ring when the ring is
-- not full. The supplied pointer is usually the end of the ring.
--
-- Unsafe because the supplied Ptr is not checked to be in range.
{-# INLINE unsafeFoldRing #-}
unsafeFoldRing :: forall a b. Storable a
    => Ptr a -> (b -> a -> b) -> b -> Ring a -> b
unsafeFoldRing ptr f z Ring{..} =
    let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->
                    go z p ptr
    in res
    where
      go !acc !p !q
        | p == q = return acc
        | otherwise = do
            x <- peek p
            go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q

-- XXX Can we remove MonadIO here?
withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
withForeignPtrM fp fn = do
    r <- fn $ unsafeForeignPtrToPtr fp
    liftIO $ touchForeignPtr fp
    return r

-- | Like unsafeFoldRing but with a monadic step function.
{-# INLINE unsafeFoldRingM #-}
unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
    => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
unsafeFoldRingM ptr f z Ring {..} =
    withForeignPtrM ringStart $ \x -> go z x ptr
  where
    go !acc !start !end
        | start == end = return acc
        | otherwise = do
            let !x = A.unsafeInlineIO $ peek start
            acc' <- f acc x
            go acc' (start `plusPtr` sizeOf (undefined :: a)) end

-- | Fold the entire length of a ring buffer starting at the supplied ringHead
-- pointer.  Assuming the supplied ringHead pointer points to the oldest item,
-- this would fold the ring starting from the oldest item to the newest item in
-- the ring.
{-# INLINE unsafeFoldRingFullM #-}
unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
    => Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
unsafeFoldRingFullM rh f z rb@Ring {..} =
    withForeignPtrM ringStart $ \_ -> go z rh
  where
    go !acc !start = do
        let !x = A.unsafeInlineIO $ peek start
        acc' <- f acc x
        let ptr = advance rb start
        if ptr == rh
            then return acc'
            else go acc' ptr

1	-- \|
2	-- Module : Streamly.Memory.Ring
3	-- Copyright : (c) 2019 Composewell Technologies
4	-- License : BSD3
5	-- Maintainer : streamly@composewell.com
6	-- Stability : experimental
7	-- Portability : GHC
8	--
9
10	module Streamly.Memory.Ring
11	( Ring(..)
12
13	-- * Construction
14	, new
15	, advance
16	, moveBy
17
18	-- * Modification
19	, unsafeInsert
20
21	-- * Folds
22	, unsafeFoldRing
23	, unsafeFoldRingM
24	, unsafeFoldRingFullM
25
26	-- * Fast Byte Comparisons
27	, unsafeEqArray
28	, unsafeEqArrayN
29	) where
30
31	import Control.Exception (assert)
32	import Foreign.ForeignPtr (ForeignPtr, withForeignPtr, touchForeignPtr)
33	import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
34	import Foreign.Ptr (plusPtr, minusPtr, castPtr)
35	import Foreign.Storable (Storable(..))
36	import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes)
37	import GHC.Ptr (Ptr(..))
38	import Prelude hiding (length, concat)
39
40	import Control.Monad.IO.Class (MonadIO(..))
41
42	import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as A
43
44	-- \| A ring buffer is a mutable array of fixed size. Initially the array is
45	-- empty, with ringStart pointing at the start of allocated memory. We call the
46	-- next location to be written in the ring as ringHead. Initially ringHead ==
47	-- ringStart. When the first item is added, ringHead points to ringStart +
48	-- sizeof item. When the buffer becomes full ringHead would wrap around to
49	-- ringStart. When the buffer is full, ringHead always points at the oldest
50	-- item in the ring and the newest item added always overwrites the oldest
51	-- item.
52	--
53	-- When using it we should keep in mind that a ringBuffer is a mutable data
54	-- structure. We should not leak out references to it for immutable use.
55	--
56	data Ring a = Ring
57	{ ringStart :: {-# UNPACK #-} !(ForeignPtr a) -- first address	×
58	, ringBound :: {-# UNPACK #-} !(Ptr a) -- first address beyond allocated memory	1✔
59	}
60
61	-- \| Create a new ringbuffer and return the ring buffer and the ringHead.
62	-- Returns the ring and the ringHead, the ringHead is same as ringStart.
63	{-# INLINE new #-}
64	new :: forall a. Storable a => Int -> IO (Ring a, Ptr a)
65	new count = do	1✔
66	let size = count * sizeOf (undefined :: a)	×
67	fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))	×
68	let p = unsafeForeignPtrToPtr fptr	1✔
69	return (Ring	1✔
70	{ ringStart = fptr	1✔
71	, ringBound = p `plusPtr` size	1✔
72	}, p)	1✔
73
74	-- \| Advance the ringHead by 1 item, wrap around if we hit the end of the
75	-- array.
76	{-# INLINE advance #-}
77	advance :: forall a. Storable a => Ring a -> Ptr a -> Ptr a
78	advance Ring{..} ringHead =	1✔
79	let ptr = ringHead `plusPtr` sizeOf (undefined :: a)	×
80	in if ptr < ringBound	1✔
81	then ptr	1✔
82	else unsafeForeignPtrToPtr ringStart	1✔
83
84	-- \| Move the ringHead by n items. The direction depends on the sign on whether
85	-- n is positive or negative. Wrap around if we hit the beginning or end of the
86	-- array.
87	{-# INLINE moveBy #-}
88	moveBy :: forall a. Storable a => Int -> Ring a -> Ptr a -> Ptr a
89	moveBy by Ring {..} ringHead = ringStartPtr `plusPtr` advanceFromHead	×
90
91	where
92
93	elemSize = sizeOf (undefined :: a)	×
94	ringStartPtr = unsafeForeignPtrToPtr ringStart	×
95	lenInBytes = ringBound `minusPtr` ringStartPtr	×
96	offInBytes = ringHead `minusPtr` ringStartPtr	×
97	len = assert (lenInBytes `mod` elemSize == 0) $ lenInBytes `div` elemSize	×
98	off = assert (offInBytes `mod` elemSize == 0) $ offInBytes `div` elemSize	×
99	advanceFromHead = (off + by `mod` len) * elemSize	×
100
101	-- \| Insert an item at the head of the ring, when the ring is full this
102	-- replaces the oldest item in the ring with the new item. This is unsafe
103	-- beause ringHead supplied is not verified to be within the Ring. Also,
104	-- the ringStart foreignPtr must be guaranteed to be alive by the caller.
105	{-# INLINE unsafeInsert #-}
106	unsafeInsert :: Storable a => Ring a -> Ptr a -> a -> IO (Ptr a)
107	unsafeInsert rb ringHead newVal = do	1✔
108	poke ringHead newVal	1✔
109	-- touchForeignPtr (ringStart rb)
110	return $ advance rb ringHead	1✔
111
112	-- XXX remove all usage of A.unsafeInlineIO
113	--
114	-- \| Like 'unsafeEqArray' but compares only N bytes instead of entire length of
115	-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to
116	-- be in range.
117	{-# INLINE unsafeEqArrayN #-}
118	unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool
119	unsafeEqArrayN Ring{..} rh A.Array{..} n =	×
120	let !res = A.unsafeInlineIO $ do	×
121	let rs = unsafeForeignPtrToPtr ringStart	×
122	as = unsafeForeignPtrToPtr aStart	×
123	assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())	×
124	let len = ringBound `minusPtr` rh	×
125	r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)	×
126	r2 <- if n > len	×
127	then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))	×
128	(min (rh `minusPtr` rs) (n - len))	×
129	else return True	×
130	-- XXX enable these, check perf impact
131	-- touchForeignPtr ringStart
132	-- touchForeignPtr aStart
133	return (r1 && r2)	×
134	in res
135
136	-- \| Byte compare the entire length of ringBuffer with the given array,
137	-- starting at the supplied ringHead pointer. Returns true if the Array and
138	-- the ringBuffer have identical contents.
139	--
140	-- This is unsafe because the ringHead Ptr is not checked to be in range. The
141	-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS
142	-- ARE NOT CHECKED.
143	{-# INLINE unsafeEqArray #-}
144	unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool
145	unsafeEqArray Ring{..} rh A.Array{..} =	×
146	let !res = A.unsafeInlineIO $ do	×
147	let rs = unsafeForeignPtrToPtr ringStart	×
148	let as = unsafeForeignPtrToPtr aStart	×
149	assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)	×
150	(return ())	×
151	let len = ringBound `minusPtr` rh	×
152	r1 <- A.memcmp (castPtr rh) (castPtr as) len	×
153	r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))	×
154	(rh `minusPtr` rs)	×
155	-- XXX enable these, check perf impact
156	-- touchForeignPtr ringStart
157	-- touchForeignPtr aStart
158	return (r1 && r2)	×
159	in res
160
161	-- XXX use MonadIO
162	--
163	-- \| Fold the buffer starting from ringStart up to the given 'Ptr' using a pure
164	-- step function. This is useful to fold the items in the ring when the ring is
165	-- not full. The supplied pointer is usually the end of the ring.
166	--
167	-- Unsafe because the supplied Ptr is not checked to be in range.
168	{-# INLINE unsafeFoldRing #-}
169	unsafeFoldRing :: forall a b. Storable a
170	=> Ptr a -> (b -> a -> b) -> b -> Ring a -> b
171	unsafeFoldRing ptr f z Ring{..} =	1✔
172	let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->	1✔
173	go z p ptr	1✔
174	in res
175	where
176	go !acc !p !q	1✔
177	\| p == q = return acc	1✔
178	\| otherwise = do	1✔
179	x <- peek p	1✔
180	go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q	×
181
182	-- XXX Can we remove MonadIO here?
183	withForeignPtrM :: MonadIO m => ForeignPtr a -> (Ptr a -> m b) -> m b
184	withForeignPtrM fp fn = do	1✔
185	r <- fn $ unsafeForeignPtrToPtr fp	1✔
186	liftIO $ touchForeignPtr fp	1✔
187	return r	1✔
188
189	-- \| Like unsafeFoldRing but with a monadic step function.
190	{-# INLINE unsafeFoldRingM #-}
191	unsafeFoldRingM :: forall m a b. (MonadIO m, Storable a)
192	=> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
193	unsafeFoldRingM ptr f z Ring {..} =	1✔
194	withForeignPtrM ringStart $ \x -> go z x ptr	1✔
195	where
196	go !acc !start !end	1✔
197	\| start == end = return acc	1✔
198	\| otherwise = do	1✔
199	let !x = A.unsafeInlineIO $ peek start	1✔
200	acc' <- f acc x	1✔
201	go acc' (start `plusPtr` sizeOf (undefined :: a)) end	×
202
203	-- \| Fold the entire length of a ring buffer starting at the supplied ringHead
204	-- pointer. Assuming the supplied ringHead pointer points to the oldest item,
205	-- this would fold the ring starting from the oldest item to the newest item in
206	-- the ring.
207	{-# INLINE unsafeFoldRingFullM #-}
208	unsafeFoldRingFullM :: forall m a b. (MonadIO m, Storable a)
209	=> Ptr a -> (b -> a -> m b) -> b -> Ring a -> m b
210	unsafeFoldRingFullM rh f z rb@Ring {..} =	1✔
211	withForeignPtrM ringStart $ \_ -> go z rh	1✔
212	where
213	go !acc !start = do	1✔
214	let !x = A.unsafeInlineIO $ peek start	1✔
215	acc' <- f acc x	1✔
216	let ptr = advance rb start	1✔
217	if ptr == rh	1✔
218	then return acc'	1✔
219	else go acc' ptr	1✔

composewell / streamly / 3375

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous