c9232b0 · streamly · git.iko.soy

benchmark/Streamly/Benchmark/FileSystem/Handle/ReadWrite.hs+1−1

compare with master

unifiedsplit

@@ -24,7 +24,7 @@ where

24

25import System.IO (Handle)

26import Prelude hiding (last, length)

27−import Streamly.Internal.Data.Array.Foreign.Type (defaultChunkSize)

27+import Streamly.Internal.System.IO (defaultChunkSize)

28

29import qualified Streamly.FileSystem.Handle as FH

30import qualified Streamly.Internal.Data.Unfold as IUF

src/Streamly/Internal/Data/Array/Foreign.hs+3−2

@@ -144,6 +144,7 @@ import Streamly.Internal.Data.Producer.Type (Producer)

144import Streamly.Internal.Data.Stream.Serial (SerialT)

145import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

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

147+import Streamly.Internal.System.IO (unsafeInlineIO)

147

148

148import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA

149import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA

149import qualified Streamly.Internal.Data.Array.Foreign.Type as A

150import qualified Streamly.Internal.Data.Array.Foreign.Type as A

@@ -236,7 +237,7 @@ unsafeRead = Unfold step inject

236 --

237 --

237 -- This should be safe as the array contents are guaranteed to be

238 -- This should be safe as the array contents are guaranteed to be

238 -- evaluated/written to before we peek at them.

239 -- evaluated/written to before we peek at them.

239− let !x = A.unsafeInlineIO $ do

240+ let !x = unsafeInlineIO $ do

240 r <- peek (Ptr p)

241 r <- peek (Ptr p)

241 touch contents

242 touch contents

242 return r

243 return r

@@ -378,7 +379,7 @@ getIndex :: Storable a => Array a -> Int -> Maybe a

378getIndex arr i =

379getIndex arr i =

379 if i < 0 || i > length arr - 1

380 if i < 0 || i > length arr - 1

380 then Nothing

381 then Nothing

381− else A.unsafeInlineIO $

382+ else unsafeInlineIO $

382 unsafeWithForeignPtr (aStart arr) $ \p -> Just <$> peekElemOff p i

383 unsafeWithForeignPtr (aStart arr) $ \p -> Just <$> peekElemOff p i

383

384

384{-

385{-

src/Streamly/Internal/Data/Array/Foreign/Mut/Type.hs+37−61

@@ -1,5 +1,3 @@

1−{-# LANGUAGE UnboxedTuples #-}

2−

3-- |

1-- |

4-- Module : Streamly.Internal.Data.Array.Foreign.Mut.Type

2-- Module : Streamly.Internal.Data.Array.Foreign.Mut.Type

@@ -92,13 +90,10 @@ module Streamly.Internal.Data.Array.Foreign.Mut.Type

92 , writeChunks

90 , writeChunks

93

91

94 -- * Utilities

92 -- * Utilities

95− , defaultChunkSize

96− , mkChunkSize

97− , mkChunkSizeKB

98 , bytesToElemCount

93 , bytesToElemCount

99− , unsafeInlineIO

100 , memcpy

94 , memcpy

101 , memcmp

95 , memcmp

96+ , c_memchr

102 )

97 )

103where

98where

104

99

@@ -118,10 +113,10 @@ import Foreign.ForeignPtr (touchForeignPtr)

118import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

113import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

119import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)

114import Foreign.Ptr (plusPtr, minusPtr, castPtr, nullPtr)

120import Foreign.Storable (Storable(..))

115import Foreign.Storable (Storable(..))

121−import GHC.Base (nullAddr#, realWorld#, build)

116+import GHC.Base (nullAddr#, build)

122import GHC.Exts (IsList, IsString(..))

117import GHC.Exts (IsList, IsString(..))

123import GHC.ForeignPtr (ForeignPtr(..))

118import GHC.ForeignPtr (ForeignPtr(..))

124−import GHC.IO (IO(IO), unsafePerformIO)

119+import GHC.IO (unsafePerformIO)

125import GHC.Ptr (Ptr(..))

120import GHC.Ptr (Ptr(..))

126

121

127import Streamly.Internal.BaseCompat

122import Streamly.Internal.BaseCompat

@@ -131,6 +126,9 @@ import Streamly.Internal.Data.SVar (adaptState)

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

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

132import Text.Read (readPrec, readListPrec, readListPrecDefault)

127import Text.Read (readPrec, readListPrec, readListPrecDefault)

133

128

129+import Streamly.Internal.System.IO

130+ (defaultChunkSize, mkChunkSize, unsafeInlineIO)

131+

134#ifdef DEVBUILD

132#ifdef DEVBUILD

135import qualified Data.Foldable as F

133import qualified Data.Foldable as F

136#endif

134#endif

@@ -156,6 +154,37 @@ plusForeignPtr (ForeignPtr addr c) (I# d) = ForeignPtr (plusAddr# addr d) c

156-- Array Data Type

154-- Array Data Type

157-------------------------------------------------------------------------------

155-------------------------------------------------------------------------------

158

156

157+foreign import ccall unsafe "string.h memcpy" c_memcpy

158+ :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)

159+

160+foreign import ccall unsafe "string.h memchr" c_memchr

161+ :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)

162+

163+foreign import ccall unsafe "string.h memcmp" c_memcmp

164+ :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt

165+

166+{-# INLINE bytesToElemCount #-}

167+bytesToElemCount :: Storable a => a -> Int -> Int

168+bytesToElemCount x n =

169+ let elemSize = sizeOf x

170+ in n + elemSize - 1 `div` elemSize

171+

172+-- XXX we are converting Int to CSize

173+memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()

174+memcpy dst src len = void (c_memcpy dst src (fromIntegral len))

175+

176+-- XXX we are converting Int to CSize

177+-- return True if the memory locations have identical contents

178+{-# INLINE memcmp #-}

179+memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool

180+memcmp p1 p2 len = do

181+ r <- c_memcmp p1 p2 (fromIntegral len)

182+ return $ r == 0

183+

184+-------------------------------------------------------------------------------

185+-- Array Data Type

186+-------------------------------------------------------------------------------

187+

159-- $arrayNotes

188-- $arrayNotes

160--

189--

161-- We can use a 'Storable' constraint in the Array type and the constraint can

190-- We can use a 'Storable' constraint in the Array type and the constraint can

@@ -183,59 +212,6 @@ mutableArray ::

183 ForeignPtr a -> Ptr a -> Ptr a -> Array a

212 ForeignPtr a -> Ptr a -> Ptr a -> Array a

184mutableArray = Array

213mutableArray = Array

185

214

186−-------------------------------------------------------------------------------

187−-- Utility functions

188−-------------------------------------------------------------------------------

189−

190−foreign import ccall unsafe "string.h memcpy" c_memcpy

191− :: Ptr Word8 -> Ptr Word8 -> CSize -> IO (Ptr Word8)

192−

193−foreign import ccall unsafe "string.h memchr" c_memchr

194− :: Ptr Word8 -> Word8 -> CSize -> IO (Ptr Word8)

195−

196−-- XXX we are converting Int to CSize

197−memcpy :: Ptr Word8 -> Ptr Word8 -> Int -> IO ()

198−memcpy dst src len = void (c_memcpy dst src (fromIntegral len))

199−

200−foreign import ccall unsafe "string.h memcmp" c_memcmp

201− :: Ptr Word8 -> Ptr Word8 -> CSize -> IO CInt

202−

203−-- XXX we are converting Int to CSize

204−-- return True if the memory locations have identical contents

205−{-# INLINE memcmp #-}

206−memcmp :: Ptr Word8 -> Ptr Word8 -> Int -> IO Bool

207−memcmp p1 p2 len = do

208− r <- c_memcmp p1 p2 (fromIntegral len)

209− return $ r == 0

210−

211−{-# INLINE unsafeInlineIO #-}

212−unsafeInlineIO :: IO a -> a

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

214−

215−{-# INLINE bytesToElemCount #-}

216−bytesToElemCount :: Storable a => a -> Int -> Int

217−bytesToElemCount x n =

218− let elemSize = sizeOf x

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

220−

221−

222−-- | GHC memory management allocation header overhead

223−allocOverhead :: Int

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

225−

226−mkChunkSize :: Int -> Int

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

228−

229−mkChunkSizeKB :: Int -> Int

230−mkChunkSizeKB n = mkChunkSize (n * k)

231− where k = 1024

232−

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

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

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

236−defaultChunkSize :: Int

237−defaultChunkSize = mkChunkSizeKB 32

238−

239-------------------------------------------------------------------------------

215-------------------------------------------------------------------------------

240-- Construction

216-- Construction

241-------------------------------------------------------------------------------

217-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Array/Foreign/Type.hs+6−14

@@ -69,16 +69,6 @@ module Streamly.Internal.Data.Array.Foreign.Type

69 , bufferChunks

70 , flattenArrays

71 , flattenArraysRev

72−

73− -- * Utilities

74− , MA.defaultChunkSize

75− , MA.mkChunkSize

76− , MA.mkChunkSizeKB

77− , MA.unsafeInlineIO

78−

79− , MA.memcpy

80− , MA.memcmp

81− , MA.bytesToElemCount

82 )

72 )

83where

73where

84

74

@@ -109,6 +99,8 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

109import qualified Streamly.Internal.Data.Unfold.Type as Unfold

99import qualified Streamly.Internal.Data.Unfold.Type as Unfold

110import qualified GHC.Exts as Exts

100import qualified GHC.Exts as Exts

111

101

102+import Streamly.Internal.System.IO (unsafeInlineIO, defaultChunkSize)

103+

112#if __GLASGOW_HASKELL__ < 808

104#if __GLASGOW_HASKELL__ < 808

113import Data.Semigroup (Semigroup(..))

105import Data.Semigroup (Semigroup(..))

114#endif

106#endif

@@ -211,7 +203,7 @@ fromPtr ::

211 Storable a =>

203 Storable a =>

212#endif

204#endif

213 Int -> Ptr a -> Array a

205 Int -> Ptr a -> Array a

214−fromPtr n ptr = MA.unsafeInlineIO $ do

206+fromPtr n ptr = unsafeInlineIO $ do

215 fptr <- newForeignPtr_ ptr

207 fptr <- newForeignPtr_ ptr

216 let end = ptr `plusPtr` n

208 let end = ptr `plusPtr` n

217 return $ Array

209 return $ Array

@@ -288,7 +280,7 @@ fromAddr# n addr# = fromPtr n (castPtr $ Ptr addr#)

288fromCString# :: Addr# -> Array Word8

280fromCString# :: Addr# -> Array Word8

289fromCString# addr# = do

281fromCString# addr# = do

290 let cstr = Ptr addr#

282 let cstr = Ptr addr#

291− len = MA.unsafeInlineIO $ c_strlen cstr

283+ len = unsafeInlineIO $ c_strlen cstr

292 fromPtr (fromIntegral len) (castPtr cstr)

284 fromPtr (fromIntegral len) (castPtr cstr)

293

285

294-- | Create an 'Array' from the first N elements of a list. The array is

286-- | Create an 'Array' from the first N elements of a list. The array is

@@ -328,7 +320,7 @@ fromStreamD str = unsafeFreeze <$> MA.fromStreamD str

328{-# INLINE bufferChunks #-}

320{-# INLINE bufferChunks #-}

329bufferChunks :: (MonadIO m, Storable a) =>

321bufferChunks :: (MonadIO m, Storable a) =>

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

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

331−bufferChunks m = D.foldr K.cons K.nil $ arraysOf MA.defaultChunkSize m

323+bufferChunks m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m

332

324

333-- | @arraysOf n stream@ groups the input stream into a stream of

325-- | @arraysOf n stream@ groups the input stream into a stream of

334-- arrays of size n.

326-- arrays of size n.

@@ -381,7 +373,7 @@ unsafeIndexIO arr = MA.unsafeIndexIO (unsafeThaw arr)

381-- | Return element at the specified index without checking the bounds.

373-- | Return element at the specified index without checking the bounds.

382{-# INLINE_NORMAL unsafeIndex #-}

374{-# INLINE_NORMAL unsafeIndex #-}

383unsafeIndex :: forall a. Storable a => Array a -> Int -> a

375unsafeIndex :: forall a. Storable a => Array a -> Int -> a

384−unsafeIndex arr i = let !r = MA.unsafeInlineIO $ unsafeIndexIO arr i in r

376+unsafeIndex arr i = let !r = unsafeInlineIO $ unsafeIndexIO arr i in r

385

377

386-- | /O(1)/ Get the byte length of the array.

378-- | /O(1)/ Get the byte length of the array.

387--

379--

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

@@ -81,6 +81,8 @@ import Streamly.Internal.Data.Parser (ParseError(..))

81import Streamly.Internal.Data.Stream.Serial (SerialT)

82import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)

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

84+import Streamly.Internal.Data.Array.Foreign.Mut.Type (memcpy, bytesToElemCount)

85+import Streamly.Internal.System.IO (mkChunkSizeKB)

84

86

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

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

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

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

@@ -454,7 +456,7 @@ spliceArraysLenUnsafe len buffered = do

454 writeArr dst (MA.Array as ae _) =

456 writeArr dst (MA.Array as ae _) =

455 liftIO $ unsafeWithForeignPtr as $ \src -> do

457 liftIO $ unsafeWithForeignPtr as $ \src -> do

456 let count = ae `minusPtr` src

458 let count = ae `minusPtr` src

457− A.memcpy (castPtr dst) (castPtr src) count

459+ memcpy (castPtr dst) (castPtr src) count

458 return $ dst `plusPtr` count

460 return $ dst `plusPtr` count

459

461

460{-# INLINE _spliceArrays #-}

462{-# INLINE _spliceArrays #-}

@@ -472,7 +474,7 @@ _spliceArrays s = do

472 writeArr dst (Array as ae) =

474 writeArr dst (Array as ae) =

473 liftIO $ unsafeWithForeignPtr as $ \src -> do

475 liftIO $ unsafeWithForeignPtr as $ \src -> do

474 let count = ae `minusPtr` src

476 let count = ae `minusPtr` src

475− A.memcpy (castPtr dst) (castPtr src) count

477+ memcpy (castPtr dst) (castPtr src) count

476 return $ dst `plusPtr` count

478 return $ dst `plusPtr` count

477

479

478{-# INLINE _spliceArraysBuffered #-}

480{-# INLINE _spliceArraysBuffered #-}

@@ -487,8 +489,8 @@ _spliceArraysBuffered s = do

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

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

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

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

489spliceArraysRealloced s = do

491spliceArraysRealloced s = do

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

492+ let idst = liftIO $ MA.newArray (bytesToElemCount (undefined :: a)

491− (A.mkChunkSizeKB 4))

493+ (mkChunkSizeKB 4))

492

494

493 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)

495 arr <- S.foldlM' MA.spliceWithDoubling idst (S.map A.unsafeThaw s)

494 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr

496 liftIO $ A.unsafeFreeze <$> MA.shrinkToFit arr

src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+2−1

@@ -138,6 +138,7 @@ import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

138import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

139import Streamly.Internal.Data.Time.Units

140 (TimeUnit64, toRelTime64, diffAbsTime64)

141+import Streamly.Internal.System.IO (defaultChunkSize)

141

142

142import qualified Streamly.Internal.Data.Array.Foreign.Type as A

143import qualified Streamly.Internal.Data.Array.Foreign.Type as A

143import qualified Streamly.Internal.Data.Fold.Type as FL

144import qualified Streamly.Internal.Data.Fold.Type as FL

@@ -1133,7 +1134,7 @@ reverse' =

1133 . fromStreamK

1134 . fromStreamK

1134 . K.reverse

1135 . K.reverse

1135 . toStreamK

1136 . toStreamK

1136− . A.arraysOf A.defaultChunkSize

1137+ . A.arraysOf defaultChunkSize

1137

1138

1138------------------------------------------------------------------------------

1139------------------------------------------------------------------------------

1139-- Position Indexing

1140-- Position Indexing

src/Streamly/Internal/FileSystem/FD.hs+3−2

@@ -130,9 +130,10 @@ import Prelude hiding (read)

130import qualified GHC.IO.FD as FD

131import qualified GHC.IO.Device as RawIO

132

133−import Streamly.Internal.Data.Array.Foreign.Type (Array(..), byteLength, defaultChunkSize, unsafeFreeze)

133+import Streamly.Internal.Data.Array.Foreign.Type

134+ (Array(..), byteLength, unsafeFreeze)

134import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)

135import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)

135−

136+import Streamly.Internal.System.IO (defaultChunkSize)

136import Streamly.Internal.Data.Stream.Serial (SerialT)

137import Streamly.Internal.Data.Stream.Serial (SerialT)

137import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)

138import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)

138

139

src/Streamly/Internal/FileSystem/File.hs+2−2

@@ -96,13 +96,13 @@ import qualified System.IO as SIO

96

97import Streamly.Internal.Data.Fold.Type (Fold(..))

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

99−import Streamly.Internal.Data.Array.Foreign.Type

99+import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)

100− (Array(..), defaultChunkSize, writeNUnsafe)

101import Streamly.Internal.Data.Stream.Serial (SerialT)

100import Streamly.Internal.Data.Stream.Serial (SerialT)

102import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)

101import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)

103import Streamly.Internal.Data.SVar (MonadAsync)

102import Streamly.Internal.Data.SVar (MonadAsync)

104-- import Streamly.Data.Fold (Fold)

103-- import Streamly.Data.Fold (Fold)

105-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

104-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

105+import Streamly.Internal.System.IO (defaultChunkSize)

106

107import qualified Streamly.Internal.Data.Fold.Type as FL

108import qualified Streamly.Internal.Data.Unfold as UF

src/Streamly/Internal/FileSystem/Handle.hs+2−1

@@ -108,12 +108,13 @@ import Streamly.Data.Fold (Fold)

108import Streamly.Internal.Data.Fold.Type (Fold2(..))

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

110import Streamly.Internal.Data.Array.Foreign.Type

111− (Array(..), writeNUnsafe, defaultChunkSize , unsafeFreezeWithShrink)

111+ (Array(..), writeNUnsafe, unsafeFreezeWithShrink)

112import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)

113import Streamly.Internal.Data.Stream.Serial (SerialT)

114import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)

115import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)

116-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

117+import Streamly.Internal.System.IO (defaultChunkSize)

117

118

118import qualified Streamly.Data.Fold as FL

119import qualified Streamly.Data.Fold as FL

119import qualified Streamly.Internal.Data.Fold.Type as FL

120import qualified Streamly.Internal.Data.Fold.Type as FL

src/Streamly/Internal/Network/Inet/TCP.hs+2−1

@@ -103,7 +103,7 @@ import Network.Socket

103 socket)

104import Prelude hiding (read)

105

106−import Streamly.Internal.Data.Array.Foreign.Type (Array(..), defaultChunkSize, writeNUnsafe)

106+import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)

107import Streamly.Internal.Data.Fold.Type (Fold(..))

108import Streamly.Internal.Data.Stream.IsStream (MonadAsync)

109import Streamly.Internal.Data.SVar (fork)

@@ -112,6 +112,7 @@ import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)

112import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

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

114import Streamly.Internal.Network.Socket (SockSpec(..), accept, connections)

115+import Streamly.Internal.System.IO (defaultChunkSize)

115

116

116import qualified Control.Monad.Catch as MC

117import qualified Control.Monad.Catch as MC

117import qualified Network.Socket as Net

118import qualified Network.Socket as Net

src/Streamly/Internal/Network/Socket.hs+12−11

@@ -95,6 +95,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)

95import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)

96import Streamly.Data.Fold (Fold)

97-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

98+import Streamly.Internal.System.IO (defaultChunkSize)

98

99

99import qualified Streamly.Data.Fold as FL

100import qualified Streamly.Data.Fold as FL

100import qualified Streamly.Internal.Data.Unfold as UF

101import qualified Streamly.Internal.Data.Unfold as UF

@@ -360,7 +361,7 @@ toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())

360-- @since 0.7.0

361-- @since 0.7.0

361{-# INLINE toChunks #-}

362{-# INLINE toChunks #-}

362toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

363toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

363−toChunks = toChunksWithBufferOf A.defaultChunkSize

364+toChunks = toChunksWithBufferOf defaultChunkSize

364

365

365-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

366-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

366-- Read requests to the socket are performed using a buffer of size @bufsize@.

367-- Read requests to the socket are performed using a buffer of size @bufsize@.

@@ -389,7 +390,7 @@ readChunksWithBufferOf = Unfold step return

389-- @since 0.7.0

390-- @since 0.7.0

390{-# INLINE readChunks #-}

391{-# INLINE readChunks #-}

391readChunks :: MonadIO m => Unfold m Socket (Array Word8)

392readChunks :: MonadIO m => Unfold m Socket (Array Word8)

392−readChunks = UF.supplyFirst A.defaultChunkSize readChunksWithBufferOf

393+readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf

393

394

394-------------------------------------------------------------------------------

395-------------------------------------------------------------------------------

395-- Read File to Stream

396-- Read File to Stream

@@ -412,7 +413,7 @@ readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

412-- TODO

413-- TODO

413-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

414-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

414--

415--

415−-- > read = 'readByChunks' A.defaultChunkSize

416+-- > read = 'readByChunks' defaultChunkSize

416-- | Generate a stream of elements of the given type from a socket. The

417-- | Generate a stream of elements of the given type from a socket. The

417-- stream ends when EOF is encountered.

418-- stream ends when EOF is encountered.

418--

419--

@@ -436,7 +437,7 @@ readWithBufferOf = UF.many readChunksWithBufferOf A.read

436-- @since 0.7.0

437-- @since 0.7.0

437{-# INLINE read #-}

438{-# INLINE read #-}

438read :: MonadIO m => Unfold m Socket Word8

439read :: MonadIO m => Unfold m Socket Word8

439−read = UF.supplyFirst A.defaultChunkSize readWithBufferOf

440+read = UF.supplyFirst defaultChunkSize readWithBufferOf

440

441

441-------------------------------------------------------------------------------

442-------------------------------------------------------------------------------

442-- Writing

443-- Writing

@@ -495,33 +496,33 @@ putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m

495writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

496writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

496writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)

497writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)

497

498

498−-- > write = 'writeWithBufferOf' A.defaultChunkSize

499+-- > write = 'writeWithBufferOf' defaultChunkSize

499--

500--

500-- | Write a byte stream to a file handle. Combines the bytes in chunks of size

501-- | Write a byte stream to a file handle. Combines the bytes in chunks of size

501−-- up to 'A.defaultChunkSize' before writing. Note that the write behavior

502+-- up to 'defaultChunkSize' before writing. Note that the write behavior

502-- depends on the 'IOMode' and the current seek position of the handle.

503-- depends on the 'IOMode' and the current seek position of the handle.

503--

504--

504-- @since 0.7.0

505-- @since 0.7.0

505{-# INLINE putBytes #-}

506{-# INLINE putBytes #-}

506putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

507putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

507−putBytes = putBytesWithBufferOf A.defaultChunkSize

508+putBytes = putBytesWithBufferOf defaultChunkSize

508

509

509-- | Write a byte stream to a socket. Accumulates the input in chunks of

510-- | Write a byte stream to a socket. Accumulates the input in chunks of

510−-- up to 'A.defaultChunkSize' bytes before writing.

511+-- up to 'defaultChunkSize' bytes before writing.

511--

512--

512-- @

513-- @

513−-- write = 'writeWithBufferOf' 'A.defaultChunkSize'

514+-- write = 'writeWithBufferOf' 'defaultChunkSize'

514-- @

515-- @

515--

516--

516-- @since 0.7.0

517-- @since 0.7.0

517{-# INLINE write #-}

518{-# INLINE write #-}

518write :: MonadIO m => Socket -> Fold m Word8 ()

519write :: MonadIO m => Socket -> Fold m Word8 ()

519−write = writeWithBufferOf A.defaultChunkSize

520+write = writeWithBufferOf defaultChunkSize

520

521

521{-

522{-

522{-# INLINE write #-}

523{-# INLINE write #-}

523write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

524write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

524−write = toHandleWith A.defaultChunkSize

525+write = toHandleWith defaultChunkSize

525-}

526-}

526

527

527-------------------------------------------------------------------------------

528-------------------------------------------------------------------------------

src/Streamly/Internal/Ring/Foreign.hs+13−11

@@ -40,6 +40,8 @@ import GHC.Ptr (Ptr(..))

40import Prelude hiding (length, concat)

41

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

43+import Streamly.Internal.Data.Array.Foreign.Mut.Type (memcmp)

44+import Streamly.Internal.System.IO (unsafeInlineIO)

43

45

44import qualified Streamly.Internal.Data.Array.Foreign.Type as A

46import qualified Streamly.Internal.Data.Array.Foreign.Type as A

45

47

@@ -115,7 +117,7 @@ unsafeInsert rb ringHead newVal = do

115 -- touchForeignPtr (ringStart rb)

117 -- touchForeignPtr (ringStart rb)

116 return $ advance rb ringHead

118 return $ advance rb ringHead

117

119

118−-- XXX remove all usage of A.unsafeInlineIO

120+-- XXX remove all usage of unsafeInlineIO

119--

121--

120-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of

122-- | Like 'unsafeEqArray' but compares only N bytes instead of entire length of

121-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to

123-- the ring buffer. This is unsafe because the ringHead Ptr is not checked to

@@ -123,14 +125,14 @@ unsafeInsert rb ringHead newVal = do

123{-# INLINE unsafeEqArrayN #-}

125{-# INLINE unsafeEqArrayN #-}

124unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool

126unsafeEqArrayN :: Ring a -> Ptr a -> A.Array a -> Int -> Bool

125unsafeEqArrayN Ring{..} rh A.Array{..} n =

127unsafeEqArrayN Ring{..} rh A.Array{..} n =

126− let !res = A.unsafeInlineIO $ do

128+ let !res = unsafeInlineIO $ do

127 let rs = unsafeForeignPtrToPtr ringStart

129 let rs = unsafeForeignPtrToPtr ringStart

128 as = unsafeForeignPtrToPtr aStart

130 as = unsafeForeignPtrToPtr aStart

129 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())

131 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs) (return ())

130 let len = ringBound `minusPtr` rh

132 let len = ringBound `minusPtr` rh

131− r1 <- A.memcmp (castPtr rh) (castPtr as) (min len n)

133+ r1 <- memcmp (castPtr rh) (castPtr as) (min len n)

132 r2 <- if n > len

134 r2 <- if n > len

133− then A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))

135+ then memcmp (castPtr rs) (castPtr (as `plusPtr` len))

134 (min (rh `minusPtr` rs) (n - len))

136 (min (rh `minusPtr` rs) (n - len))

135 else return True

137 else return True

136 -- XXX enable these, check perf impact

138 -- XXX enable these, check perf impact

@@ -149,14 +151,14 @@ unsafeEqArrayN Ring{..} rh A.Array{..} n =

149{-# INLINE unsafeEqArray #-}

151{-# INLINE unsafeEqArray #-}

150unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool

152unsafeEqArray :: Ring a -> Ptr a -> A.Array a -> Bool

151unsafeEqArray Ring{..} rh A.Array{..} =

153unsafeEqArray Ring{..} rh A.Array{..} =

152− let !res = A.unsafeInlineIO $ do

154+ let !res = unsafeInlineIO $ do

153 let rs = unsafeForeignPtrToPtr ringStart

155 let rs = unsafeForeignPtrToPtr ringStart

154 let as = unsafeForeignPtrToPtr aStart

156 let as = unsafeForeignPtrToPtr aStart

155 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)

157 assert (aEnd `minusPtr` as >= ringBound `minusPtr` rs)

156 (return ())

158 (return ())

157 let len = ringBound `minusPtr` rh

159 let len = ringBound `minusPtr` rh

158− r1 <- A.memcmp (castPtr rh) (castPtr as) len

160+ r1 <- memcmp (castPtr rh) (castPtr as) len

159− r2 <- A.memcmp (castPtr rs) (castPtr (as `plusPtr` len))

161+ r2 <- memcmp (castPtr rs) (castPtr (as `plusPtr` len))

160 (rh `minusPtr` rs)

162 (rh `minusPtr` rs)

161 -- XXX enable these, check perf impact

163 -- XXX enable these, check perf impact

162 -- touchForeignPtr ringStart

164 -- touchForeignPtr ringStart

@@ -175,7 +177,7 @@ unsafeEqArray Ring{..} rh A.Array{..} =

175unsafeFoldRing :: forall a b. Storable a

177unsafeFoldRing :: forall a b. Storable a

176 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b

178 => Ptr a -> (b -> a -> b) -> b -> Ring a -> b

177unsafeFoldRing ptr f z Ring{..} =

179unsafeFoldRing ptr f z Ring{..} =

178− let !res = A.unsafeInlineIO $ withForeignPtr ringStart $ \p ->

180+ let !res = unsafeInlineIO $ withForeignPtr ringStart $ \p ->

179 go z p ptr

181 go z p ptr

180 in res

182 in res

181 where

183 where

@@ -202,7 +204,7 @@ unsafeFoldRingM ptr f z Ring {..} =

202 go !acc !start !end

204 go !acc !start !end

203 | start == end = return acc

205 | start == end = return acc

204 | otherwise = do

206 | otherwise = do

205− let !x = A.unsafeInlineIO $ peek start

207+ let !x = unsafeInlineIO $ peek start

206 acc' <- f acc x

208 acc' <- f acc x

207 go acc' (start `plusPtr` sizeOf (undefined :: a)) end

209 go acc' (start `plusPtr` sizeOf (undefined :: a)) end

208

210

@@ -220,7 +222,7 @@ unsafeFoldRingFullM rh f z rb@Ring {..} =

220 withForeignPtrM ringStart $ \_ -> go z rh

222 withForeignPtrM ringStart $ \_ -> go z rh

221 where

223 where

222 go !acc !start = do

224 go !acc !start = do

223− let !x = A.unsafeInlineIO $ peek start

225+ let !x = unsafeInlineIO $ peek start

224 acc' <- f acc x

226 acc' <- f acc x

225 let ptr = advance rb start

227 let ptr = advance rb start

226 if ptr == rh

228 if ptr == rh

@@ -242,7 +244,7 @@ unsafeFoldRingNM count rh f z rb@Ring {..} =

242

244

243 go 0 acc _ = return acc

245 go 0 acc _ = return acc

244 go !n !acc !start = do

246 go !n !acc !start = do

245− let !x = A.unsafeInlineIO $ peek start

247+ let !x = unsafeInlineIO $ peek start

246 acc' <- f acc x

248 acc' <- f acc x

247 let ptr = advance rb start

249 let ptr = advance rb start

248 if ptr == rh || n == 0

250 if ptr == rh || n == 0

src/Streamly/Internal/System/IO.hsnew file+52

@@ -0,0 +1,52 @@

1+{-# LANGUAGE UnboxedTuples #-}

2+

3+-- |

4+-- Module : Streamly.Internal.System.IO

6+-- License : BSD-3-Clause

7+-- Maintainer : streamly@composewell.com

8+-- Stability : experimental

9+-- Portability : GHC

10+--

11+

12+module Streamly.Internal.System.IO

13+ ( defaultChunkSize

14+ , mkChunkSize

15+ , mkChunkSizeKB

16+ , unsafeInlineIO

17+ )

18+

19+where

20+

21+-------------------------------------------------------------------------------

22+-- Imports

23+-------------------------------------------------------------------------------

24+

25+import Foreign.Storable (Storable(sizeOf))

26+import GHC.Base (realWorld#)

27+import GHC.IO (IO(IO))

28+

29+-------------------------------------------------------------------------------

30+-- API

31+-------------------------------------------------------------------------------

32+

33+{-# INLINE unsafeInlineIO #-}

34+unsafeInlineIO :: IO a -> a

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

36+

37+-- | GHC memory management allocation header overhead

38+allocOverhead :: Int

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

40+

41+mkChunkSize :: Int -> Int

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

43+

44+mkChunkSizeKB :: Int -> Int

45+mkChunkSizeKB n = mkChunkSize (n * k)

46+ where k = 1024

47+

48+-- | 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

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

51+defaultChunkSize :: Int

52+defaultChunkSize = mkChunkSizeKB 32

src/Streamly/Internal/Unicode/Stream.hs+2−1

@@ -98,6 +98,7 @@ import Streamly.Internal.Data.Stream.IsStream (IsStream)

98import Streamly.Internal.Data.Stream.Serial (SerialT)

99import Streamly.Internal.Data.Stream.StreamD (Stream(..), Step (..))

100import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

101+import Streamly.Internal.System.IO (unsafeInlineIO)

101

102

102import qualified Streamly.Internal.Data.Unfold as Unfold

103import qualified Streamly.Internal.Data.Unfold as Unfold

103import qualified Streamly.Internal.Data.Stream.Serial as Serial

104import qualified Streamly.Internal.Data.Stream.Serial as Serial

@@ -236,7 +237,7 @@ utf8d =

236-- and without touching the foreign ptr.

237-- and without touching the foreign ptr.

237{-# INLINE_NORMAL unsafePeekElemOff #-}

238{-# INLINE_NORMAL unsafePeekElemOff #-}

238unsafePeekElemOff :: forall a. Storable a => Ptr a -> Int -> a

239unsafePeekElemOff :: forall a. Storable a => Ptr a -> Int -> a

239−unsafePeekElemOff p i = let !x = A.unsafeInlineIO $ peekElemOff p i in x

240+unsafePeekElemOff p i = let !x = unsafeInlineIO $ peekElemOff p i in x

240

241

241-- decode is split into two separate cases to avoid branching instructions.

242-- decode is split into two separate cases to avoid branching instructions.

242-- From the higher level flow we already know which case we are in so we can

243-- From the higher level flow we already know which case we are in so we can

streamly.cabal+1

@@ -369,6 +369,7 @@ library

369 , Streamly.Internal.Data.Time.Units

370 , Streamly.Internal.Data.Time.Clock.Type

371 , Streamly.Internal.Data.Time.Clock

372+ , Streamly.Internal.System.IO

372

373

373 -- streamly-core-stream

374 -- streamly-core-stream

374 , Streamly.Internal.Data.SVar

375 , Streamly.Internal.Data.SVar

test/Streamly/Test/Network/Inet/TCP.hs+5−5

@@ -15,11 +15,11 @@ import Data.Function ((&))

15import Data.Word (Word8)

16import Network.Socket (Socket, PortNumber)

17import Streamly.Internal.Control.Monad (discard)

18+import Streamly.Internal.System.IO (defaultChunkSize)

18import Streamly.Prelude (SerialT)

19import Streamly.Prelude (SerialT)

19import Test.QuickCheck (Property)

20import Test.QuickCheck (Property)

20import Test.QuickCheck.Monadic (monadicIO, assert, run)

21import Test.QuickCheck.Monadic (monadicIO, assert, run)

21

22

22−import qualified Streamly.Internal.Data.Array.Foreign.Type as Array

23import qualified Streamly.Internal.Data.Unfold as Unfold

24import qualified Streamly.Internal.Network.Inet.TCP as TCP

25import qualified Streamly.Internal.Network.Socket as Socket

@@ -106,8 +106,8 @@ validateOnPort :: Property

106validateOnPort = monadicIO $ do

107 res <- run $ do

108 ls2 <-

109− execute TCP.acceptOnPort basePort Array.defaultChunkSize handlerRW

109+ execute TCP.acceptOnPort basePort defaultChunkSize handlerRW

110− let dataChunk = take Array.defaultChunkSize testDataSource

110+ let dataChunk = take defaultChunkSize testDataSource

111 Stream.eqBy (==) (Stream.fromList dataChunk) ls2

112 assert res

113

@@ -118,9 +118,9 @@ validateOnPortLocal = monadicIO $ do

118 execute

119 TCP.acceptOnPortLocal

120 (basePort + 1)

121− Array.defaultChunkSize

121+ defaultChunkSize

122 handlerRW

123− let dataChunk = take Array.defaultChunkSize testDataSource

123+ let dataChunk = take defaultChunkSize testDataSource

124 Stream.eqBy (==) (Stream.fromList dataChunk) ls2

125 assert res

126

test/Streamly/Test/Network/Socket.hs+3−3

@@ -15,11 +15,11 @@ import Data.Function ((&))

15import Data.Word (Word8)

16import Network.Socket (Socket, PortNumber)

17import Streamly.Internal.Control.Monad (discard)

18+import Streamly.Internal.System.IO (defaultChunkSize)

18import Streamly.Prelude (SerialT)

19import Streamly.Prelude (SerialT)

19import Test.QuickCheck (Property)

20import Test.QuickCheck (Property)

20import Test.QuickCheck.Monadic (monadicIO, assert, run)

21import Test.QuickCheck.Monadic (monadicIO, assert, run)

21

22

22−import qualified Streamly.Internal.Data.Array.Foreign.Type as Array

23import qualified Streamly.Internal.Network.Inet.TCP as TCP

24import qualified Streamly.Internal.Network.Socket as Socket

25import qualified Streamly.Internal.Unicode.Stream as Unicode

@@ -118,8 +118,8 @@ validateWithBufferOf = monadicIO $ do

118validateRW :: Property

119validateRW = monadicIO $ do

120 res <- run $ do

121− ls2 <- execute (basePort + 1) Array.defaultChunkSize handlerRW

121+ ls2 <- execute (basePort + 1) defaultChunkSize handlerRW

122− let dataChunk = take Array.defaultChunkSize testDataSource

122+ let dataChunk = take defaultChunkSize testDataSource

123 Stream.eqBy (==) (Stream.fromList dataChunk) ls2

124 assert res

125