a25ad4a · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Ensure minimum chunk size of one element

a25ad4aHarendra Kumar5 years ago

Changelog.md+1

compare with master

unifiedsplit

@@ -16,6 +16,7 @@

16* Fix a bug in classifySessionsBy, see PR #1311. The bug could cause

17 premature ejection of a session when nput events with the same key are

18 split into multiple sessions.

19+* Fix array: unnecessary additional allocation due to a bug

20## 0.8.0 (Jun 2021)

21## 0.8.0 (Jun 2021)

src/Streamly/Internal/Data/Array/Foreign/Mut/Type.hs+36−17

@@ -119,7 +119,8 @@ module Streamly.Internal.Data.Array.Foreign.Mut.Type

119

120 -- * Memory Management

121 , blockSize

122− , arrayChunkSize

122+ , arrayChunkBytes

123+ , allocBytesToElemCount

123 , realloc

124 , realloc

124 , resize

125 , resize

125 , resizeExp

126 , resizeExp

@@ -187,7 +188,6 @@ module Streamly.Internal.Data.Array.Foreign.Mut.Type

187 -- , appendSliceFrom

188 -- , appendSliceFrom

188

189

189 -- * Utilities

190 -- * Utilities

190− , bytesToElemCount

191 , memcpy

192 , memcmp

193 , c_memchr

@@ -227,7 +227,7 @@ import Streamly.Internal.Data.SVar.Type (adaptState)

227import Streamly.Internal.Data.Unfold.Type (Unfold(..))

228

229import Streamly.Internal.System.IO

230− (defaultChunkSize, mkChunkSize, unsafeInlineIO)

230+ (defaultChunkSize, arrayPayloadSize, unsafeInlineIO)

231

232#ifdef DEVBUILD

233import qualified Data.Foldable as F

@@ -272,13 +272,13 @@ foreign import ccall unsafe "string.h memcmp" c_memcmp

272 :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt

273

274-- | Given a 'Storable' type (unused first arg) and a number of bytes, return

275−-- how many elements of that type will fit in after rounding the bytes up to

275+-- how many elements of that type will completely fit in those bytes.

276−-- the element size.

276+--

277{-# INLINE bytesToElemCount #-}

278bytesToElemCount :: Storable a => a -> Int -> Int

279bytesToElemCount x n =

280 let elemSize = sizeOf x

281− in n + elemSize - 1 `div` elemSize

281+ in n `div` elemSize

282

283-- XXX we are converting Int to CSize

284memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()

@@ -588,13 +588,30 @@ roundUpToPower2 :: Int -> Int

588roundUpToPower2 = undefined

589-}

590

591−-- XXX remove mkChunkSize, it is not required.

591+-- | @allocBytesToBytes elem allocatedBytes@ returns the array size in bytes

592+-- such that the real allocation is less than or equal to @allocatedBytes@,

593+-- unless @allocatedBytes@ is less than the size of one array element in which

594+-- case it returns one element's size.

595+--

596+{-# INLINE allocBytesToBytes #-}

597+allocBytesToBytes :: forall a. Storable a => a -> Int -> Int

598+allocBytesToBytes _ n =

599+ max (arrayPayloadSize n) (sizeOf (undefined :: a))

600+

601+-- | Given a 'Storable' type (unused first arg) and real allocation size

602+-- (including overhead), return how many elements of that type will completely

603+-- fit in it, returns at least 1.

604+--

605+{-# INLINE allocBytesToElemCount #-}

606+allocBytesToElemCount :: Storable a => a -> Int -> Int

607+allocBytesToElemCount x bytes =

608+ let n = bytesToElemCount x (allocBytesToBytes x bytes)

609+ in assert (n >= 1) n

610+

592-- | The default chunk size by which the array creation routines increase the

611-- | The default chunk size by which the array creation routines increase the

593-- size of the array when the array is grown linearly.

612-- size of the array when the array is grown linearly.

594−--

613+arrayChunkBytes :: Int

595−{-# INLINE arrayChunkSize #-}

614+arrayChunkBytes = 1024

596−arrayChunkSize :: Int

597−arrayChunkSize = mkChunkSize 1024

598

615

599-------------------------------------------------------------------------------

616-------------------------------------------------------------------------------

600-- Snoc

617-- Snoc

@@ -708,7 +725,7 @@ snocWith allocSize arr x = liftIO $ do

708

725

709-- | The array is mutated to append an additional element to it. If there

726-- | The array is mutated to append an additional element to it. If there

710-- is no reserved space available in the array then it is reallocated to grow

727-- is no reserved space available in the array then it is reallocated to grow

711−-- it by 'arrayChunkSize' rounded up to 'blockSize' when the size becomes more

728+-- it by 'arrayChunkBytes' rounded up to 'blockSize' when the size becomes more

712-- than 'largeObjectThreshold'.

729-- than 'largeObjectThreshold'.

713--

730--

714-- Note that the returned array may be a mutated version of the original array.

731-- Note that the returned array may be a mutated version of the original array.

@@ -718,7 +735,7 @@ snocWith allocSize arr x = liftIO $ do

718-- /Pre-release/

735-- /Pre-release/

719{-# INLINE snocLinear #-}

736{-# INLINE snocLinear #-}

720snocLinear :: forall m a. (MonadIO m, Storable a) => Array a -> a -> m (Array a)

737snocLinear :: forall m a. (MonadIO m, Storable a) => Array a -> a -> m (Array a)

721−snocLinear = snocWith (+ arrayChunkSize)

738+snocLinear = snocWith (+ allocBytesToBytes (undefined :: a) arrayChunkBytes)

722

739

723-- XXX round it to next power of 2.

740-- XXX round it to next power of 2.

724--

741--

@@ -1162,9 +1179,11 @@ arraysOf n (D.Stream step state) =

1162-- XXX buffer to a list instead?

1179-- XXX buffer to a list instead?

1163-- | Buffer the stream into arrays in memory.

1180-- | Buffer the stream into arrays in memory.

1164{-# INLINE arrayStreamKFromStreamD #-}

1181{-# INLINE arrayStreamKFromStreamD #-}

1165−arrayStreamKFromStreamD :: (MonadIO m, Storable a) =>

1182+arrayStreamKFromStreamD :: forall m a. (MonadIO m, Storable a) =>

1166 D.Stream m a -> m (K.Stream m (Array a))

1183 D.Stream m a -> m (K.Stream m (Array a))

1167−arrayStreamKFromStreamD m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m

1184+arrayStreamKFromStreamD =

1185+ let n = allocBytesToElemCount (undefined :: a) defaultChunkSize

1186+ in D.foldr K.cons K.nil . arraysOf n

1168

1187

1169-------------------------------------------------------------------------------

1188-------------------------------------------------------------------------------

1170-- Streams of arrays - Flattening

1189-- Streams of arrays - Flattening

@@ -1725,7 +1744,7 @@ writeWith elemCount =

1725

1744

1726-- | Fold the whole input to a single array.

1745-- | Fold the whole input to a single array.

1727--

1746--

1728−-- Same as 'writeWith' using an initial array size of 'arrayChunkSize' bytes

1747+-- Same as 'writeWith' using an initial array size of 'arrayChunkBytes' bytes

1729-- rounded up to the element size.

1748-- rounded up to the element size.

1730--

1749--

1731-- /Caution! Do not use this on infinite streams./

1750-- /Caution! Do not use this on infinite streams./

@@ -1733,7 +1752,7 @@ writeWith elemCount =

1733-- @since 0.7.0

1752-- @since 0.7.0

1734{-# INLINE write #-}

1753{-# INLINE write #-}

1735write :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)

1754write :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)

1736−write = writeWith (bytesToElemCount (undefined :: a) arrayChunkSize)

1755+write = writeWith (allocBytesToElemCount (undefined :: a) arrayChunkBytes)

1737

1756

1738-------------------------------------------------------------------------------

1757-------------------------------------------------------------------------------

1739-- construct from streams, known size

1758-- construct from streams, known size

src/Streamly/Internal/Data/Array/Stream/Foreign.hs+4−4

@@ -75,8 +75,8 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)

75import Streamly.Internal.Data.Stream.IsStream.Type

76 (IsStream, fromStreamD, toStreamD)

77import Streamly.Internal.Data.SVar (adaptState, defState)

78−import Streamly.Internal.Data.Array.Foreign.Mut.Type (memcpy, bytesToElemCount)

78+import Streamly.Internal.Data.Array.Foreign.Mut.Type

79−import Streamly.Internal.System.IO (mkChunkSizeKB)

79+ (memcpy, allocBytesToElemCount)

81import qualified Streamly.Internal.Data.Array.Foreign as A

82import qualified Streamly.Internal.Data.Array.Foreign as Array

@@ -470,8 +470,8 @@ _spliceArraysBuffered s = do

470spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)

471 => SerialT m (Array a) -> m (Array a)

472spliceArraysRealloced s = do

473+ let n = allocBytesToElemCount (undefined :: a) (4 * 1024)

473− let idst = liftIO $ MA.newArray (bytesToElemCount (undefined :: a)

474+ idst = liftIO $ MA.newArray n

474− (mkChunkSizeKB 4))

475

476 arr <- S.foldlM' MA.spliceExp idst (S.map A.unsafeThaw s)

477 liftIO $ A.unsafeFreeze <$> MA.rightSize arr

src/Streamly/Internal/System/IO.hs+16−12

@@ -11,8 +11,7 @@

12module Streamly.Internal.System.IO

13 ( defaultChunkSize

14− , mkChunkSize

14+ , arrayPayloadSize

15− , mkChunkSizeKB

16 , unsafeInlineIO

15 , unsafeInlineIO

17 )

16 )

@@ -34,19 +33,24 @@ import GHC.IO (IO(IO))

34unsafeInlineIO :: IO a -> a

33unsafeInlineIO :: IO a -> a

35unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r

34unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r

37−-- | GHC memory management allocation header overhead

36+-- | Returns the heap allocation overhead for allocating a byte array. Each

38−allocOverhead :: Int

37+-- heap object contains a one word header. Byte arrays contain the size of the

38+-- array after the header.

39+--

40+-- See https://gitlab.haskell.org/ghc/ghc/-/wikis/commentary/rts/storage/heap-objects#arrays

41+--

42+byteArrayOverhead :: Int

39−allocOverhead = 2 * sizeOf (undefined :: Int)

43+byteArrayOverhead = 2 * sizeOf (undefined :: Word)

40−

41−mkChunkSize :: Int -> Int

42−mkChunkSize n = let size = n - allocOverhead in max size 0

44−mkChunkSizeKB :: Int -> Int

45+-- | When we allocate a byte array of size @k@ the allocator actually allocates

45−mkChunkSizeKB n = mkChunkSize (n * k)

46+-- memory of size @k + byteArrayOverhead@. @arrayPayloadSize n@ returns the

46− where k = 1024

47+-- size of the array in bytes that would result in an allocation of @n@ bytes.

48+--

49+arrayPayloadSize :: Int -> Int

50+arrayPayloadSize n = let size = n - byteArrayOverhead in max size 0

48-- | Default maximum buffer size in bytes, for reading from and writing to IO

52-- | Default maximum buffer size in bytes, for reading from and writing to IO

49-- devices, the value is 32KB minus GHC allocation overhead, which is a few

53-- devices, the value is 32KB minus GHC allocation overhead, which is a few

50-- bytes, so that the actual allocation is 32KB.

54-- bytes, so that the actual allocation is 32KB.

51defaultChunkSize :: Int

55defaultChunkSize :: Int

52−defaultChunkSize = mkChunkSizeKB 32

56+defaultChunkSize = arrayPayloadSize (32 * 1024)