7336d58 · streamly · git.iko.soy

git.iko.soy/streamly

Fix a bug caused by unsafeInlineIO

unsafe use of unsafeInlineIO caused each array allocated in the toArrayN fold to share the same memory. This fix uses the IO monad to make sure that the code is not pure and therefore we always allocate new memory. All such usage of unsafePerformIo have been fixed. The remaining ones are reviewed to be safe.

commit7336d585a97e06f5fb5ff6ebadd78c1913340930 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 12 May 2019, 20:38 (7 years ago) parent6cd235f

12 files changed+257−194

compare with master

unifiedsplit

benchmark/Array.hs+40−22 benchmark/ArrayOps.hs+30−27 benchmark/LinearOps.hs+4−3 benchmark/NanoBenchmarks.hs+4−4 src/Streamly/Array.hs+9−11 src/Streamly/Array/Types.hs+56−40 src/Streamly/Fold.hs+4−3 src/Streamly/Prelude.hs+1−1 src/Streamly/RingBuffer.hs+41−30 src/Streamly/Streams/StreamD.hs+31−30 test/Arrays.hs+36−23 test/Prop.hs+1

benchmark/Array.hs+40−22

@@ -7,15 +7,14 @@

77-- Maintainer : harendra.kumar@gmail.com

99import Control.DeepSeq (NFData(..), deepseq)

10−import Data.Functor.Identity (Identity)

1110import Foreign.Storable (Storable(..))

1211import System.Random (randomRIO)

1312

14−import Streamly

1513import qualified GHC.Exts as GHC

1614

1715import qualified ArrayOps as Ops

1816import qualified Streamly.Array as A

17+import qualified Streamly.Prelude as S

1918

2019import Gauge

2120

@@ -28,15 +27,36 @@ benchPure :: NFData b => String -> (Int -> a) -> (a -> b) -> Benchmark

2827benchPure name src f = bench name $ nfIO $

2928 randomRIO (1,1) >>= return . f . src

3029

31−{-# INLINE benchPureSink #-}

32−benchPureSink :: NFData b => String -> (Ops.Stream Int -> b) -> Benchmark

33−benchPureSink name f = benchPure name Ops.sourceIntFromTo f

34−

30+-- Drain a source that generates a pure array

3531{-# INLINE benchPureSrc #-}

3632benchPureSrc :: (NFData a, Storable a)

3733 => String -> (Int -> Ops.Stream a) -> Benchmark

3834benchPureSrc name src = benchPure name src id

3935

36+{-# INLINE benchIO #-}

37+benchIO :: NFData b => String -> (Int -> IO a) -> (a -> b) -> Benchmark

38+benchIO name src f = bench name $ nfIO $

39+ randomRIO (1,1) >>= src >>= return . f

40+

41+-- Drain a source that generates an array in the IO monad

42+{-# INLINE benchIOSrc #-}

43+benchIOSrc :: (NFData a, Storable a)

44+ => String -> (Int -> IO (Ops.Stream a)) -> Benchmark

45+benchIOSrc name src = benchIO name src id

46+

47+{-# INLINE benchPureSink #-}

48+benchPureSink :: NFData b => String -> (Ops.Stream Int -> b) -> Benchmark

49+benchPureSink name f = benchIO name Ops.sourceIntFromTo f

50+

51+{-# INLINE benchIO' #-}

52+benchIO' :: NFData b => String -> (Int -> IO a) -> (a -> IO b) -> Benchmark

53+benchIO' name src f = bench name $ nfIO $

54+ randomRIO (1,1) >>= src >>= f

55+

56+{-# INLINE benchIOSink #-}

57+benchIOSink :: NFData b => String -> (Ops.Stream Int -> IO b) -> Benchmark

58+benchIOSink name f = benchIO' name Ops.sourceIntFromTo f

59+

4060mkString :: String

4161mkString = "[1" ++ concat (replicate Ops.value ",1") ++ "]"

4262

@@ -45,11 +65,11 @@ main =

4565 defaultMain

4666 [ bgroup "array"

4767 [ bgroup "generation"

48− [ benchPureSrc "fromStreamN . intFromTo" Ops.sourceIntFromTo

49− , benchPureSrc "fromStream . intFromTo" Ops.sourceIntFromToFromStream

50− , benchPureSrc "fromList . intFromTo" Ops.sourceIntFromToFromList

51− , benchPureSrc "fromStreamN . unfoldr" Ops.sourceUnfoldr

52− , benchPureSrc "fromStreamN . fromList" Ops.sourceFromList

68+ [ benchIOSrc "fromStreamN . intFromTo" Ops.sourceIntFromTo

69+ , benchIOSrc "fromStream . intFromTo" Ops.sourceIntFromToFromStream

70+ , benchIOSrc "fromList . intFromTo" Ops.sourceIntFromToFromList

71+ , benchIOSrc "fromStreamN . unfoldr" Ops.sourceUnfoldr

72+ , benchIOSrc "fromStreamN . fromList" Ops.sourceFromList

5373 , benchPureSrc "fromStreamN . IsList.fromList" Ops.sourceIsList

5474 , benchPureSrc "fromStreamN . IsString.fromString" Ops.sourceIsString

5575 , mkString `deepseq` (bench "read" $ nf Ops.readInstance mkString)

@@ -66,11 +86,9 @@ main =

6686 , benchPureSink "<" Ops.ordInstance

6787 , benchPureSink "min" Ops.ordInstanceMin

6888 , benchPureSink "IsList.toList" GHC.toList

69− , benchPureSink "foldl'" Ops.pureFoldl'

70− , benchPureSink "toStream"

71− (A.toStream :: A.Array Int -> SerialT Identity Int)

72− , benchPureSink "toStreamRev"

73− (A.toStreamRev :: A.Array Int -> SerialT Identity Int)

89+ , benchIOSink "foldl'" Ops.pureFoldl'

90+ , benchIOSink "toStream" (S.drain . A.toStream)

91+ , benchIOSink "toStreamRev" (S.drain . A.toStreamRev)

7492#ifdef DEVBUILD

7593 , benchPureSink "foldable/foldl'" Ops.foldableFoldl'

7694 , benchPureSink "foldable/sum" Ops.foldableSum

@@ -118,9 +136,9 @@ main =

118136 ]

119137 -}

120138 , bgroup "transformation"

121− [ benchPureSink "scanl'" (Ops.scanl' 1)

122− , benchPureSink "scanl1'" (Ops.scanl1' 1)

123− , benchPureSink "map" (Ops.map 1)

139+ [ benchIOSink "scanl'" (Ops.scanl' 1)

140+ , benchIOSink "scanl1'" (Ops.scanl1' 1)

141+ , benchIOSink "map" (Ops.map 1)

124142 {-

125143 , benchPureSink "fmap" (Ops.fmap 1)

126144 , benchPureSink "mapM" (Ops.mapM serially 1)

@@ -138,9 +156,9 @@ main =

138156 -}

139157 ]

140158 , bgroup "transformationX4"

141− [ benchPureSink "scanl'" (Ops.scanl' 4)

142− , benchPureSink "scanl1'" (Ops.scanl1' 4)

143− , benchPureSink "map" (Ops.map 4)

159+ [ benchIOSink "scanl'" (Ops.scanl' 4)

160+ , benchIOSink "scanl1'" (Ops.scanl1' 4)

161+ , benchIOSink "map" (Ops.map 4)

144162 {-

145163 , benchPureSink "fmap" (Ops.fmap 4)

146164 , benchPureSink "mapM" (Ops.mapM serially 4)

benchmark/ArrayOps.hs+30−27

@@ -14,7 +14,7 @@

1414module ArrayOps where

1515

1616-- import Control.Monad (when)

17−import Data.Functor.Identity (Identity, runIdentity)

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

1818-- import Data.Maybe (fromJust)

1919import Prelude (Int, Bool, (+), ($), (==), (>), (.), Maybe(..), undefined)

2020import qualified Prelude as P

@@ -48,8 +48,8 @@ maxValue = value + 1

4848type Stream = A.Array

4949

5050{-# INLINE sourceUnfoldr #-}

51−sourceUnfoldr :: Int -> Stream Int

52−sourceUnfoldr n = runIdentity $ A.fromStreamN value $ S.unfoldr step n

51+sourceUnfoldr :: MonadIO m => Int -> m (Stream Int)

52+sourceUnfoldr n = A.fromStreamN value $ S.unfoldr step n

5353 where

5454 step cnt =

5555 if cnt > n + value

@@ -57,30 +57,27 @@ sourceUnfoldr n = runIdentity $ A.fromStreamN value $ S.unfoldr step n

5757 else (Just (cnt, cnt + 1))

5858

5959{-# INLINE sourceIntFromTo #-}

60−sourceIntFromTo :: Int -> Stream Int

61−sourceIntFromTo n = runIdentity $

62− A.fromStreamN value $ S.enumerateFromTo n (n + value)

60+sourceIntFromTo :: MonadIO m => Int -> m (Stream Int)

61+sourceIntFromTo n = A.fromStreamN value $ S.enumerateFromTo n (n + value)

6362

6463{-# INLINE sourceIntFromToFromStream #-}

65−sourceIntFromToFromStream :: Int -> Stream Int

66−sourceIntFromToFromStream n = runIdentity $

67− A.fromStream $ S.enumerateFromTo n (n + value)

64+sourceIntFromToFromStream :: MonadIO m => Int -> m (Stream Int)

65+sourceIntFromToFromStream n = A.fromStream $ S.enumerateFromTo n (n + value)

6866

69−sourceIntFromToFromList :: Int -> Stream Int

70−sourceIntFromToFromList n = A.fromList $ [n..n + value]

67+sourceIntFromToFromList :: MonadIO m => Int -> m (Stream Int)

68+sourceIntFromToFromList n = P.return $ A.fromList $ [n..n + value]

7169

7270{-# INLINE sourceFromList #-}

73−sourceFromList :: Int -> Stream Int

74−sourceFromList n = runIdentity $ A.fromStreamN value $ S.fromList [n..n+value]

71+sourceFromList :: MonadIO m => Int -> m (Stream Int)

72+sourceFromList n = A.fromStreamN value $ S.fromList [n..n+value]

7573

7674{-# INLINE sourceIsList #-}

7775sourceIsList :: Int -> Stream Int

78−sourceIsList n = runIdentity $ A.fromStreamN value $ GHC.fromList [n..n+value]

76+sourceIsList n = GHC.fromList [n..n+value]

7977

8078{-# INLINE sourceIsString #-}

8179sourceIsString :: Int -> Stream P.Char

82−sourceIsString n = runIdentity $

83− A.fromStreamN value $ GHC.fromString (P.replicate (n + value) 'a')

80+sourceIsString n = GHC.fromString (P.replicate (n + value) 'a')

8481

8582{-

8683-------------------------------------------------------------------------------

@@ -209,18 +206,21 @@ maximumBy = S.maximumBy compare

209206-- Transformation

210207-------------------------------------------------------------------------------

211208

209+{-

212210{-# INLINE transform #-}

213211transform :: Stream a -> Stream a

214212transform = P.id

213+-}

215214

216215{-# INLINE composeN #-}

217−composeN :: Int -> (Stream Int -> Stream Int) -> Stream Int -> Stream Int

218−composeN n f =

216+composeN :: P.Monad m

217+ => Int -> (Stream Int -> m (Stream Int)) -> Stream Int -> m (Stream Int)

218+composeN n f x =

219219 case n of

220− 1 -> transform . f

221− 2 -> transform . f . f

222− 3 -> transform . f . f . f

223− 4 -> transform . f . f . f . f

220+ 1 -> f x

221+ 2 -> f x P.>>= f

222+ 3 -> f x P.>>= f P.>>= f

223+ 4 -> f x P.>>= f P.>>= f P.>>= f

224224 _ -> undefined

225225

226226{-# INLINE scanl' #-}

@@ -256,7 +256,7 @@ scanl' , scanl1', map{-, fmap, mapMaybe, filterEven, filterAllOut,

256256 dropAll, dropWhileTrue, dropWhileMTrue, dropWhileFalse,

257257 findIndices, elemIndices, insertBy, deleteBy, reverse,

258258 foldrS, foldrSMap, foldrT, foldrTMap -}

259− :: Int -> Stream Int -> Stream Int

259+ :: MonadIO m => Int -> Stream Int -> m (Stream Int)

260260

261261{-

262262{-# INLINE mapMaybeM #-}

@@ -276,8 +276,11 @@ sequence :: (S.IsStream t, S.MonadAsync m)

276276 => (t m Int -> S.SerialT m Int) -> t m (m Int) -> m ()

277277 -}

278278

279−onArray :: (S.SerialT Identity Int -> S.SerialT Identity Int) -> Stream Int -> Stream Int

280−onArray f arr = runIdentity $ A.fromStreamN value $ f $ A.toStream arr

279+onArray

280+ :: MonadIO m => (S.SerialT m Int -> S.SerialT m Int)

281+ -> Stream Int

282+ -> m (Stream Int)

283+onArray f arr = A.fromStreamN value $ f $ A.toStream arr

281284

282285scanl' n = composeN n $ onArray $ S.scanl' (+) 0

283286scanl1' n = composeN n $ onArray $ S.scanl1' (+)

@@ -507,8 +510,8 @@ readInstance str =

507510 _ -> P.error "readInstance: no parse"

508511

509512{-# INLINE pureFoldl' #-}

510−pureFoldl' :: Stream Int -> Int

511−pureFoldl' = runIdentity . S.foldl' (+) 0 . A.toStream

513+pureFoldl' :: MonadIO m => Stream Int -> m Int

514+pureFoldl' = S.foldl' (+) 0 . A.toStream

512515

513516#ifdef DEVBUILD

514517{-# INLINE foldableFoldl' #-}

benchmark/LinearOps.hs+4−3

@@ -14,6 +14,7 @@

1414module LinearOps where

1515

1616import Control.Monad (when)

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

1718import Data.Functor.Identity (Identity, runIdentity)

1819import Data.Maybe (fromJust)

1920import Prelude

@@ -287,7 +288,7 @@ transform = runStream

287288

288289{-# INLINE composeN #-}

289290composeN

290− :: Monad m

291+ :: MonadIO m

291292 => Int -> (Stream m Int -> Stream m Int) -> Stream m Int -> m ()

292293composeN n f =

293294 case n of

@@ -342,7 +343,7 @@ scan, scanl1', map, fmap, mapMaybe, filterEven, filterAllOut,

342343 dropAll, dropWhileTrue, dropWhileMTrue, dropWhileFalse,

343344 findIndices, elemIndices, insertBy, deleteBy, reverse, reverse',

344345 foldrS, foldrSMap, foldrT, foldrTMap

345− :: Monad m

346+ :: MonadIO m

346347 => Int -> Stream m Int -> m ()

347348

348349{-# INLINE mapMaybeM #-}

@@ -529,7 +530,7 @@ concatMap n = S.concatMap (\_ -> sourceUnfoldrMN maxNested n)

529530{-# INLINE filterMap #-}

530531scanMap, dropMap, dropScan, takeDrop, takeScan, takeMap, filterDrop,

531532 filterTake, filterScan, filterScanl1, filterMap

532− :: Monad m => Int -> Stream m Int -> m ()

533+ :: MonadIO m => Int -> Stream m Int -> m ()

533534

534535scanMap n = composeN n $ S.map (subtract 1) . S.scanl' (+) 0

535536dropMap n = composeN n $ S.map (subtract 1) . S.drop 1

benchmark/NanoBenchmarks.hs+4−4

@@ -52,7 +52,7 @@ takeAllDropOne = S.drop 1 . S.take maxValue

5252

5353{-# INLINE takeDrop #-}

5454takeDrop :: Monad m => S.Stream m Int -> m ()

55−takeDrop = S.runStream .

55+takeDrop = S.drain .

5656 takeAllDropOne . takeAllDropOne . takeAllDropOne . takeAllDropOne

5757

5858-------------------------------------------------------------------------------

@@ -67,7 +67,7 @@ dropWhileFalse = S.dropWhile (> maxValue)

6767

6868{-# INLINE dropWhileFalseX4 #-}

6969dropWhileFalseX4 :: Monad m => S.Stream m Int -> m ()

70−dropWhileFalseX4 = S.runStream

70+dropWhileFalseX4 = S.drain

7171 . dropWhileFalse . dropWhileFalse . dropWhileFalse . dropWhileFalse

7272

7373-------------------------------------------------------------------------------

@@ -88,9 +88,9 @@ iterateSource g i n = f i (sourceUnfoldrMN n)

8888main :: IO ()

8989main = do

9090 defaultMain [bench "unfoldr" $ nfIO $

91− randomRIO (1,1) >>= \n -> S.runStream (sourceUnfoldr n)]

91+ randomRIO (1,1) >>= \n -> S.drain (sourceUnfoldr n)]

9292 defaultMain [bench "take-drop" $ nfIO $ takeDrop sourceUnfoldrM]

9393 defaultMain [bench "dropWhileFalseX4" $

9494 nfIO $ dropWhileFalseX4 sourceUnfoldrM]

9595 defaultMain [bench "iterate-mapM" $

96− nfIO $ S.runStream $ iterateSource (S.mapM return) 100000 10]

96+ nfIO $ S.drain $ iterateSource (S.mapM return) 100000 10]

src/Streamly/Array.hs+9−11

@@ -142,7 +142,6 @@ import System.IO (Handle, hGetBufSome, hPutBuf)

142142import Prelude hiding (length, null, last, map, (!!))

143143

144144import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

145−import GHC.IO (unsafeDupablePerformIO)

146145

147146import Streamly.Streams.Serial (SerialT)

148147import Streamly.Streams.StreamK.Type (IsStream, mkStream)

@@ -162,7 +161,7 @@ import qualified Streamly.Prelude as S

162161--

163162-- @since 0.7.0

164163{-# INLINE fromStreamN #-}

165−fromStreamN :: (Monad m, Storable a) => Int -> SerialT m a -> m (Array a)

164+fromStreamN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)

166165fromStreamN n m = fromStreamDN n $ D.toStreamD m

167166

168167-------------------------------------------------------------------------------

@@ -219,7 +218,7 @@ arr !! i =

219218--

220219-- @since 0.7.0

221220{-# INLINE flattenArrays #-}

222−flattenArrays :: (IsStream t, Monad m, Storable a) => t m (Array a) -> t m a

221+flattenArrays :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

223222flattenArrays m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)

224223

225224-- XXX should we have a reverseArrays API to reverse the stream of arrays

@@ -230,7 +229,7 @@ flattenArrays m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)

230229--

231230-- @since 0.7.0

232231{-# INLINE _flattenArraysRev #-}

233−_flattenArraysRev :: (IsStream t, Monad m, Storable a) => t m (Array a) -> t m a

232+_flattenArraysRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

234233_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)

235234

236235-- |

@@ -240,7 +239,7 @@ _flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)

240239--

241240-- @since 0.7.0

242241{-# INLINE arraysOf #-}

243−arraysOf :: (IsStream t, Monad m, Storable a)

242+arraysOf :: (IsStream t, MonadIO m, Storable a)

244243 => Int -> t m a -> t m (Array a)

245244arraysOf n str =

246245 D.fromStreamD $ fromStreamDArraysOf n (D.toStreamD str)

@@ -250,23 +249,22 @@ arraysOf n str =

250249--

251250-- @since 0.7.0

252251{-# INLINABLE spliceArrays #-}

253−spliceArrays :: (Monad m, Storable a) => SerialT m (Array a) -> m (Array a)

252+spliceArrays :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)

254253spliceArrays s = do

255254 buffered <- S.foldr S.cons S.nil s

256255 len <- S.sum (S.map length buffered)

257256

258− let !arr = unsafeDupablePerformIO $ unsafeNew len

259− end <- S.foldl' write (aEnd arr) buffered

257+ arr <- liftIO $ newArray len

258+ end <- S.foldlM' write (aEnd arr) buffered

260259 return $ arr {aEnd = end}

261260

262261 where

263262

264263 write dst Array{..} =

265− let !dst' = unsafeInlineIO $ withForeignPtr aStart $ \src -> do

264+ liftIO $ withForeignPtr aStart $ \src -> do

266265 let len = aEnd `minusPtr` src

267266 memcpy (castPtr dst) (castPtr src) len

268267 return $ dst `plusPtr` len

269− in dst'

270268

271269-- | Create an 'Array' from a stream. This is useful when we want to create a

272270-- single array from a stream of unknown size. 'fromStreamN' is at least twice

@@ -277,7 +275,7 @@ spliceArrays s = do

277275-- processing of indvidual arrays in the resulting stream should be preferred.

278276--

279277{-# INLINE fromStream #-}

280−fromStream :: (Monad m, Storable a) => SerialT m a -> m (Array a)

278+fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

281279fromStream m = A.fromStreamD $ D.toStreamD m

282280

283281{-

src/Streamly/Array/Types.hs+56−40

@@ -25,7 +25,7 @@ module Streamly.Array.Types

2525 -- * Construction

2626 , unsafeInlineIO

2727 , withNewArray

28− , unsafeNew

28+ , newArray

2929 , unsafeAppend

3030 , shrinkToFit

3131 , memcpy

@@ -40,6 +40,7 @@ module Streamly.Array.Types

4040 , flattenArraysRev

4141

4242 -- * Elimination

43+ , unsafeIndexIO

4344 , unsafeIndex

4445 , length

4546 , foldl'

@@ -59,6 +60,7 @@ where

5960import Control.Exception (assert)

6061import Control.DeepSeq (NFData(..))

6162import Control.Monad (when)

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

6264import Data.Functor.Identity (runIdentity)

6365import Data.Word (Word8)

6466import Foreign.C.String (CString)

@@ -74,7 +76,7 @@ import Text.Read (readPrec, readListPrec, readListPrecDefault)

7476import GHC.Base (Addr#, realWorld#)

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

7678import GHC.ForeignPtr (mallocPlainForeignPtrAlignedBytes, newForeignPtr_)

77−import GHC.IO (IO(IO), unsafeDupablePerformIO)

79+import GHC.IO (IO(IO), unsafePerformIO)

7880import GHC.Ptr (Ptr(..))

7981

8082import Streamly.Fold.Types (Fold(..))

@@ -84,6 +86,10 @@ import qualified Streamly.Streams.StreamD.Type as D

8486import qualified Streamly.Streams.StreamK as K

8587import qualified GHC.Exts as Exts

8688

89+#ifdef DEVBUILD

90+import qualified Data.Foldable as F

91+#endif

92+

8793-------------------------------------------------------------------------------

8894-- Design Notes

8995-------------------------------------------------------------------------------

@@ -178,9 +184,9 @@ unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r

178184--

179185-- Note that this is internal routine, the reference to this array cannot be

180186-- given out until the array has been written to and frozen.

181−{-# INLINE unsafeNew #-}

182−unsafeNew :: forall a. Storable a => Int -> IO (Array a)

183−unsafeNew count = do

187+{-# INLINE newArray #-}

188+newArray :: forall a. Storable a => Int -> IO (Array a)

189+newArray count = do

184190 let size = count * sizeOf (undefined :: a)

185191 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))

186192 let p = unsafeForeignPtrToPtr fptr

@@ -195,7 +201,7 @@ unsafeNew count = do

195201{-# INLINE withNewArray #-}

196202withNewArray :: forall a. Storable a => Int -> (Ptr a -> IO ()) -> IO (Array a)

197203withNewArray count f = do

198− arr <- unsafeNew count

204+ arr <- newArray count

199205 withForeignPtr (aStart arr) $ \p -> f p >> return arr

200206

201207-- XXX grow the array when we are beyond bound.

@@ -263,18 +269,23 @@ _fromCStringAddrUnsafe addr# = do

263269-- Elimination

264270-------------------------------------------------------------------------------

265271

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

273+--

274+-- Unsafe because it does not check the bounds of the array.

275+{-# INLINE_NORMAL unsafeIndexIO #-}

276+unsafeIndexIO :: forall a. Storable a => Array a -> Int -> IO a

277+unsafeIndexIO Array {..} i =

278+ withForeignPtr aStart $ \p -> do

279+ let elemSize = sizeOf (undefined :: a)

280+ elemOff = p `plusPtr` (elemSize * i)

281+ assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)

282+ (return ())

283+ peek elemOff

284+

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

267286{-# INLINE_NORMAL unsafeIndex #-}

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

269−unsafeIndex Array {..} i =

270− let !r = unsafeInlineIO $

271− withForeignPtr aStart $ \p -> do

272− let elemSize = sizeOf (undefined :: a)

273− elemOff = p `plusPtr` (elemSize * i)

274− assert (i >= 0 && elemOff `plusPtr` elemSize <= aEnd)

275− (return ())

276− peek elemOff

277− in r

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

278289

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

280291--

@@ -297,6 +308,12 @@ toStreamD Array{..} =

297308 {-# INLINE_LATE step #-}

298309 step _ p | p == aEnd = return D.Stop

299310 step _ p = do

311+ -- unsafeInlineIO allows us to run this in Identity monad for pure

312+ -- toList/foldr case which makes them much faster due to not

313+ -- accumulating the list and fusing better with the pure consumers.

314+ --

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

316+ -- evaluated/written to before we peek at them.

300317 let !x = unsafeInlineIO $ do

301318 r <- peek p

302319 touchForeignPtr aStart

@@ -314,6 +331,7 @@ toStreamDRev Array{..} =

314331 {-# INLINE_LATE step #-}

315332 step _ p | p < unsafeForeignPtrToPtr aStart = return D.Stop

316333 step _ p = do

334+ -- See comments in toStreamD for why we use unsafeInlineIO

317335 let !x = unsafeInlineIO $ do

318336 r <- peek p

319337 touchForeignPtr aStart

@@ -337,18 +355,16 @@ foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr

337355--

338356-- @since 0.7.0

339357{-# INLINE_NORMAL toArrayN #-}

340−toArrayN :: forall m a. (Monad m, Storable a) => Int -> Fold m a (Array a)

358+toArrayN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)

341359toArrayN n = Fold step initial extract

342360

343361 where

344362

345− initial =

346− let !arr = unsafeDupablePerformIO $ unsafeNew n

347− in return arr

363+ initial = liftIO $ newArray n

348364 step arr@(Array _ end bound) _ | end == bound = return arr

349− step (Array start end bound) x =

350− let !_ = unsafeInlineIO $ poke end x

351− in return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound

365+ step (Array start end bound) x = do

366+ liftIO $ poke end x

367+ return $ Array start (end `plusPtr` sizeOf (undefined :: a)) bound

352368 extract = return

353369

354370{-

@@ -366,23 +382,23 @@ toArray = Fold step begin done

366382-}

367383

368384{-# INLINE_NORMAL fromStreamDN #-}

369−fromStreamDN :: forall m a. (Monad m, Storable a)

385+fromStreamDN :: forall m a. (MonadIO m, Storable a)

370386 => Int -> D.Stream m a -> m (Array a)

371387fromStreamDN limit str = do

372− let !arr = unsafeDupablePerformIO $ unsafeNew limit

373− end <- D.foldl' write (aEnd arr) $ D.take limit str

388+ arr <- liftIO $ newArray limit

389+ end <- D.foldlM' write (aEnd arr) $ D.take limit str

374390 return $ arr {aEnd = end}

375391

376392 where

377393

378− write ptr x =

379− let !_ = unsafeInlineIO $ poke ptr x

380− in ptr `plusPtr` sizeOf (undefined :: a)

394+ write ptr x = do

395+ liftIO $ poke ptr x

396+ return $ ptr `plusPtr` sizeOf (undefined :: a)

381397

382398-- | @fromStreamArraysOf n stream@ groups the input stream into a stream of

383399-- arrays of size n.

384400{-# INLINE fromStreamDArraysOf #-}

385−fromStreamDArraysOf :: (Monad m, Storable a)

401+fromStreamDArraysOf :: (MonadIO m, Storable a)

386402 => Int -> D.Stream m a -> D.Stream m (Array a)

387403fromStreamDArraysOf n str = D.groupsOf n (toArrayN n) str

388404

@@ -393,7 +409,7 @@ data FlattenState s a =

393409 | InnerLoop s (ForeignPtr a) (Ptr a) (Ptr a)

394410

395411{-# INLINE_NORMAL flattenArrays #-}

396−flattenArrays :: forall m a. (Monad m, Storable a)

412+flattenArrays :: forall m a. (MonadIO m, Storable a)

397413 => D.Stream m (Array a) -> D.Stream m a

398414flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)

399415

@@ -413,7 +429,7 @@ flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)

413429 return $ D.Skip $ OuterLoop st

414430

415431 step' _ (InnerLoop st startf p end) = do

416− let !x = unsafeInlineIO $ do

432+ x <- liftIO $ do

417433 r <- peek p

418434 touchForeignPtr startf

419435 return r

@@ -421,7 +437,7 @@ flattenArrays (D.Stream step state) = D.Stream step' (OuterLoop state)

421437 (p `plusPtr` (sizeOf (undefined :: a))) end)

422438

423439{-# INLINE_NORMAL flattenArraysRev #-}

424−flattenArraysRev :: forall m a. (Monad m, Storable a)

440+flattenArraysRev :: forall m a. (MonadIO m, Storable a)

425441 => D.Stream m (Array a) -> D.Stream m a

426442flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)

427443

@@ -442,7 +458,7 @@ flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)

442458 return $ D.Skip $ OuterLoop st

443459

444460 step' _ (InnerLoop st startf p end) = do

445− let !x = unsafeInlineIO $ do

461+ x <- liftIO $ do

446462 r <- peek p

447463 touchForeignPtr startf

448464 return r

@@ -454,7 +470,7 @@ flattenArraysRev (D.Stream step state) = D.Stream step' (OuterLoop state)

454470-- all the arrays.

455471--

456472{-# INLINE fromStreamD #-}

457−fromStreamD :: (Monad m, Storable a) => D.Stream m a -> m (Array a)

473+fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)

458474fromStreamD m = do

459475 let s = fromStreamDArraysOf defaultChunkSize m

460476 buffered <- D.foldr K.cons K.nil s

@@ -479,14 +495,14 @@ instance (Show a, Storable a) => Show (Array a) where

479495-- @since 0.7.0

480496{-# INLINABLE fromListN #-}

481497fromListN :: Storable a => Int -> [a] -> Array a

482−fromListN n xs = runIdentity $ fromStreamDN n $ D.fromList xs

498+fromListN n xs = unsafePerformIO $ fromStreamDN n $ D.fromList xs

483499

484500-- | Create an 'Array' from a list. The list must be of finite size.

485501--

486502-- @since 0.7.0

487503{-# INLINABLE fromList #-}

488504fromList :: Storable a => [a] -> Array a

489−fromList xs = runIdentity $ fromStreamD $ D.fromList xs

505+fromList xs = unsafePerformIO $ fromStreamD $ D.fromList xs

490506

491507instance (Storable a, Read a, Show a) => Read (Array a) where

492508 {-# INLINE readPrec #-}

@@ -540,7 +556,7 @@ instance (Storable a, NFData a) => NFData (Array a) where

540556

541557instance (Storable a, Ord a) => Ord (Array a) where

542558 {-# INLINE compare #-}

543− compare arr1 arr2 = runIdentity $

559+ compare arr1 arr2 = unsafePerformIO $

544560 D.cmpBy compare (toStreamD arr1) (toStreamD arr2)

545561

546562 -- Default definitions defined in base do not have an INLINE on them, so we

@@ -570,7 +586,7 @@ instance (Storable a, Ord a) => Ord (Array a) where

570586-- make the Foldable instance possible though it is much slower (7x slower).

571587--

572588{-# INLINE_NORMAL toStreamD_ #-}

573−toStreamD_ :: forall m a. Monad m => Int -> Array a -> D.Stream m a

589+toStreamD_ :: forall m a. MonadIO m => Int -> Array a -> D.Stream m a

574590toStreamD_ size Array{..} =

575591 let p = unsafeForeignPtrToPtr aStart

576592 in D.Stream step p

@@ -580,7 +596,7 @@ toStreamD_ size Array{..} =

580596 {-# INLINE_LATE step #-}

581597 step _ p | p == aEnd = return D.Stop

582598 step _ p = do

583− let !x = unsafeInlineIO $ do

599+ x <- liftIO $ do

584600 r <- peek p

585601 touchForeignPtr aStart

586602 return r

@@ -590,7 +606,7 @@ toStreamD_ size Array{..} =

590606_foldr :: forall a b. (a -> b -> b) -> b -> Array a -> b

591607_foldr f z arr@Array {..} =

592608 let !n = sizeOf (undefined :: a)

593− in runIdentity $ D.foldr f z $ toStreamD_ n arr

609+ in unsafePerformIO $ D.foldr f z $ toStreamD_ n arr

594610

595611-- | Note that the 'Foldable' instance is 7x slower than the direct

596612-- operations.

src/Streamly/Fold.hs+4−3

@@ -294,6 +294,7 @@ module Streamly.Fold

294294 )

295295where

296296

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

297298import Data.Functor.Identity (Identity)

298299import Foreign.Storable (Storable(..))

299300import Prelude

@@ -1347,7 +1348,7 @@ _splitSuffixBy' predicate f m = grouped f (S.map (\a -> (a, predicate a)) m)

13471348-- @since 0.7.0

13481349{-# INLINE splitOn #-}

13491350splitOn

1350− :: (IsStream t, Monad m, Storable a, Enum a, Eq a)

1351+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)

13511352 => Array a -> Fold m a b -> t m a -> t m b

13521353splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)

13531354

@@ -1395,7 +1396,7 @@ splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.to

13951396-- @since 0.7.0

13961397{-# INLINE splitSuffixOn #-}

13971398splitSuffixOn

1398− :: (IsStream t, Monad m, Storable a, Enum a, Eq a)

1399+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)

13991400 => Array a -> Fold m a b -> t m a -> t m b

14001401splitSuffixOn patt f m =

14011402 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)

@@ -1482,7 +1483,7 @@ splitOn' patt f m = S.intersperseM

14821483-- @since 0.7.0

14831484{-# INLINE splitSuffixOn' #-}

14841485splitSuffixOn'

1485− :: (IsStream t, Monad m, Storable a, Enum a, Eq a)

1486+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)

14861487 => Array a -> Fold m a b -> t m a -> t m b

14871488splitSuffixOn' patt f m =

14881489 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)

src/Streamly/Prelude.hs+1−1

@@ -1875,7 +1875,7 @@ reverse s = fromStreamS $ S.reverse $ toStreamS s

18751875--

18761876-- @since 0.7.0

18771877{-# INLINE reverse' #-}

1878−reverse' :: (IsStream t, Monad m, Storable a) => t m a -> t m a

1878+reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a

18791879reverse' s = fromStreamD $ D.reverse' $ toStreamD s

18801880

18811881------------------------------------------------------------------------------

src/Streamly/RingBuffer.hs+41−30

@@ -16,15 +16,15 @@ module Streamly.RingBuffer

1616 ( RingBuffer(..)

1717

1818 -- * Construction

19− , unsafeNew

19+ , new

2020

2121 -- * Modification

22− , insert

22+ , unsafeInsert

2323

2424 -- * Folds

25− , foldRing

26− , foldRingM

27− , foldRingFullM

25+ , unsafeFoldRing

26+ , unsafeFoldRingM

27+ , unsafeFoldRingFullM

2828

2929 -- * Fast Byte Comparisons

3030 , unsafeEqArray

@@ -45,10 +45,10 @@ import qualified Streamly.Array.Types as A

4545-- | A ring buffer is a mutable array of fixed size. Initially the array is

4646-- empty, with ringStart pointing at the start of allocated memory. We call the

4747-- next location to be written in the ring as ringHead. Initially ringHead ==

48−-- ringStart. When the first item is added ringHead points to ringStart +

48+-- ringStart. When the first item is added, ringHead points to ringStart +

4949-- sizeof item. When the buffer becomes full ringHead would wrap around to

5050-- ringStart. When the buffer is full, ringHead always points at the oldest

51−-- item in the ring and the newest item added will always overwrite the oldest

51+-- item in the ring and the newest item added always overwrites the oldest

5252-- item.

5353--

5454-- When using it we should keep in mind that a ringBuffer is a mutable data

@@ -60,9 +60,10 @@ data RingBuffer a = RingBuffer

6060 }

6161

6262-- | Create a new ringbuffer and return the ring buffer and the ringHead.

63−{-# INLINE unsafeNew #-}

64−unsafeNew :: forall a. Storable a => Int -> IO (RingBuffer a, Ptr a)

65−unsafeNew count = do

63+-- Returns the ring and the ringHead, the ringHead is same as ringStart.

64+{-# INLINE new #-}

65+new :: forall a. Storable a => Int -> IO (RingBuffer a, Ptr a)

66+new count = do

6667 let size = count * sizeOf (undefined :: a)

6768 fptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: a))

6869 let p = unsafeForeignPtrToPtr fptr

@@ -82,16 +83,21 @@ advance RingBuffer{..} ringHead =

8283 else unsafeForeignPtrToPtr ringStart

8384

8485-- | Insert an item at the head of the ring, when the ring is full this

85−-- replaces the oldest item in the ring with the new item.

86−{-# INLINE insert #-}

87−insert :: Storable a => RingBuffer a -> Ptr a -> a -> IO (Ptr a)

88−insert rb ringHead newVal = do

86+-- replaces the oldest item in the ring with the new item. This is unsafe

87+-- beause ringHead supplied is not verified to be within the RingBuffer. Also,

88+-- the ringStart foreignPtr must be guaranteed to be alive by the caller.

89+{-# INLINE unsafeInsert #-}

90+unsafeInsert :: Storable a => RingBuffer a -> Ptr a -> a -> IO (Ptr a)

91+unsafeInsert rb ringHead newVal = do

8992 poke ringHead newVal

9093 -- touchForeignPtr (ringStart rb)

9194 return $ advance rb ringHead

9295

96+-- XXX remove all usage of A.unsafeInlineIO

97+--

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

94−-- the ring buffer.

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

100+-- be in range.

95101{-# INLINE unsafeEqArrayN #-}

96102unsafeEqArrayN :: RingBuffer a -> Ptr a -> A.Array a -> Int -> Bool

97103unsafeEqArrayN RingBuffer{..} rh A.Array{..} n =

@@ -105,17 +111,17 @@ unsafeEqArrayN RingBuffer{..} rh A.Array{..} n =

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

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

107113 else return True

114+ -- XXX enable these, check perf impact

108115 -- touchForeignPtr ringStart

109116 -- touchForeignPtr aStart

110117 return (r1 && r2)

111118 in res

112119

113120-- | Byte compare the entire length of ringBuffer with the given array,

114−-- starting at the supplied pointer. The supplied pointer is usually the

115−-- ringHead, therefore it would compare the ringBuffer starting at the oldest

116−-- item in the ring up to the newest item in the ring.

121+-- starting at the supplied ringHead pointer. Returns true if the Array and

122+-- the ringBuffer have identical contents.

117123--

118−-- Returns true if the Array and the ringBuffer have identical contents. The

124+-- This is unsafe because the ringHead Ptr is not checked to be in range. The

119125-- supplied array must be equal to or bigger than the ringBuffer, ARRAY BOUNDS

120126-- ARE NOT CHECKED.

121127{-# INLINE unsafeEqArray #-}

@@ -130,18 +136,23 @@ unsafeEqArray RingBuffer{..} rh A.Array{..} =

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

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

132138 (rh `minusPtr` rs)

139+ -- XXX enable these, check perf impact

133140 -- touchForeignPtr ringStart

134141 -- touchForeignPtr aStart

135142 return (r1 && r2)

136143 in res

137144

138−-- | Fold the buffer starting from ringStart to a given position using a pure

145+-- XXX use MonadIO

146+--

147+-- | Fold the buffer starting from ringStart up to the given 'Ptr' using a pure

139148-- step function. This is useful to fold the items in the ring when the ring is

140149-- not full. The supplied pointer is usually the end of the ring.

141−{-# INLINE foldRing #-}

142−foldRing :: forall a b. Storable a

150+--

151+-- Unsafe because the supplied Ptr is not checked to be in range.

152+{-# INLINE unsafeFoldRing #-}

153+unsafeFoldRing :: forall a b. Storable a

143154 => Ptr a -> (b -> a -> b) -> b -> RingBuffer a -> b

144−foldRing ptr f z RingBuffer{..} =

155+unsafeFoldRing ptr f z RingBuffer{..} =

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

146157 go z p ptr

147158 in res

@@ -152,11 +163,11 @@ foldRing ptr f z RingBuffer{..} =

152163 x <- peek p

153164 go (f acc x) (p `plusPtr` sizeOf (undefined :: a)) q

154165

155−-- | Like foldRing but with a monadic step function.

156−{-# INLINE foldRingM #-}

157−foldRingM :: forall m a b. (Monad m, Storable a)

166+-- | Like unsafeFoldRing but with a monadic step function.

167+{-# INLINE unsafeFoldRingM #-}

168+unsafeFoldRingM :: forall m a b. (Monad m, Storable a)

158169 => Ptr a -> (b -> a -> m b) -> b -> RingBuffer a -> m b

159−foldRingM ptr f z RingBuffer{..} = go z (unsafeForeignPtrToPtr ringStart) ptr

170+unsafeFoldRingM ptr f z RingBuffer{..} = go z (unsafeForeignPtrToPtr ringStart) ptr

160171 where

161172 go !acc !start !end

162173 | start == end = return acc

@@ -169,10 +180,10 @@ foldRingM ptr f z RingBuffer{..} = go z (unsafeForeignPtrToPtr ringStart) ptr

169180-- pointer. Assuming the supplied ringHead pointer points to the oldest item,

170181-- this would fold the ring starting from the oldest item to the newest item in

171182-- the ring.

172−{-# INLINE foldRingFullM #-}

173−foldRingFullM :: forall m a b. (Monad m, Storable a)

183+{-# INLINE unsafeFoldRingFullM #-}

184+unsafeFoldRingFullM :: forall m a b. (Monad m, Storable a)

174185 => Ptr a -> (b -> a -> m b) -> b -> RingBuffer a -> m b

175−foldRingFullM rh f z rb@RingBuffer{..} = go z rh

186+unsafeFoldRingFullM rh f z rb@RingBuffer{..} = go z rh

176187 where

177188 go !acc !start = do

178189 let !x = A.unsafeInlineIO $ peek start

src/Streamly/Streams/StreamD.hs+31−30

@@ -228,6 +228,7 @@ module Streamly.Streams.StreamD

228228 )

229229where

230230

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

231232import Control.Monad.Trans (MonadTrans(lift))

232233import Data.Word (Word32)

233234import Data.Bits (shiftR, shiftL, (.|.), (.&.))

@@ -241,10 +242,8 @@ import Prelude

241242 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,

242243 (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat,

243244 reverse)

244−import System.IO.Unsafe (unsafeDupablePerformIO)

245245

246−import Streamly.Array.Types

247− (Array(..), unsafeInlineIO, unsafeIndex, flattenArrays)

246+import Streamly.Array.Types (Array(..), flattenArrays)

248247import Streamly.Fold.Types (Fold(..))

249248import Streamly.SVar (MonadAsync, defState, adaptState)

250249import Streamly.Sink.Types (Sink(..))

@@ -836,7 +835,7 @@ reverse m = Stream step Nothing

836835

837836-- Much faster reverse for Storables

838837{-# INLINE_NORMAL reverse' #-}

839−reverse' :: forall m a. (Monad m, Storable a) => Stream m a -> Stream m a

838+reverse' :: forall m a. (MonadIO m, Storable a) => Stream m a -> Stream m a

840839{-

841840-- This commented implementation copies the whole stream into one single array

842841-- and then streams from that array, this is 3-4x faster than the chunked code

@@ -1148,7 +1147,7 @@ data SplitOnState s a =

11481147

11491148{-# INLINE_NORMAL splitOn #-}

11501149splitOn

1151− :: forall m a b. (Monad m, Storable a, Enum a, Eq a)

1150+ :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)

11521151 => Array a

11531152 -> Fold m a b

11541153 -> Stream m a

@@ -1167,13 +1166,14 @@ splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =

11671166 if patLen == 0

11681167 then return $ Skip $ GO_EMPTY_PAT state

11691168 else if patLen == 1

1170− then return $ Skip $ GO_SINGLE_PAT state (unsafeIndex patArr 0)

1169+ then do

1170+ r <- liftIO $ (A.unsafeIndexIO patArr 0)

1171+ return $ Skip $ GO_SINGLE_PAT state r

11711172 else if sizeOf (undefined :: a) * patLen

11721173 <= sizeOf (undefined :: Word)

11731174 then return $ Skip $ GO_SHORT_PAT state

11741175 else do

1175− let !(rb, rhead) = unsafeDupablePerformIO $

1176− RB.unsafeNew patLen

1176+ (rb, rhead) <- liftIO $ RB.new patLen

11771177 return $ Skip $ GO_KARP_RABIN state rb rhead

11781178

11791179 stepOuter gst (GO_SINGLE_PAT stt pat) = initial >>= go SPEC stt

@@ -1264,10 +1264,10 @@ splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =

12641264 res <- step (adaptState gst) st

12651265 case res of

12661266 Yield x s -> do

1267− let !rh' = unsafeInlineIO (RB.insert rb rh x)

1267+ rh' <- liftIO $ RB.unsafeInsert rb rh x

12681268 if idx == maxIndex

12691269 then do

1270− let fold = RB.foldRing (RB.ringBound rb)

1270+ let fold = RB.unsafeFoldRing (RB.ringBound rb)

12711271 let !ringHash = fold addCksum 0 rb

12721272 if ringHash == patHash

12731273 then go2 SPEC ringHash rh' s acc

@@ -1276,7 +1276,7 @@ splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =

12761276 Skip s -> go0 SPEC idx rh s acc

12771277 Stop -> do

12781278 !acc' <- if idx /= 0

1279− then RB.foldRingM rh fstep acc rb

1279+ then RB.unsafeFoldRingM rh fstep acc rb

12801280 else return acc

12811281 done acc' >>= \r -> return $ Yield r GO_DONE

12821282

@@ -1290,20 +1290,20 @@ splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =

12901290 res <- step (adaptState gst) st

12911291 case res of

12921292 Yield x s -> do

1293− let !old = unsafeInlineIO $ peek rh

1294− cksum' = deltaCksum cksum old x

1293+ old <- liftIO $ peek rh

1294+ let cksum' = deltaCksum cksum old x

12951295 acc' <- fstep acc old

12961296

12971297 if (cksum' == patHash)

12981298 then do

1299− let !rh'= unsafeInlineIO (RB.insert rb rh x)

1299+ rh' <- liftIO (RB.unsafeInsert rb rh x)

13001300 go2 SPEC cksum' rh' s acc'

13011301 else do

1302− let !rh'= unsafeInlineIO (RB.insert rb rh x)

1302+ rh' <- liftIO (RB.unsafeInsert rb rh x)

13031303 go1 SPEC cksum' rh' s acc'

13041304 Skip s -> go1 SPEC cksum rh s acc

13051305 Stop -> do

1306− acc' <- RB.foldRingFullM rh fstep acc rb

1306+ acc' <- RB.unsafeFoldRingFullM rh fstep acc rb

13071307 done acc' >>= \r -> return $ Yield r GO_DONE

13081308

13091309 go2 !_ !cksum' !rh' s !acc' = do

@@ -1327,7 +1327,7 @@ splitOn patArr@Array{..} (Fold fstep initial done) (Stream step state) =

13271327

13281328{-# INLINE_NORMAL splitSuffixOn #-}

13291329splitSuffixOn

1330− :: forall m a b. (Monad m, Storable a, Enum a, Eq a)

1330+ :: forall m a b. (MonadIO m, Storable a, Enum a, Eq a)

13311331 => Bool

13321332 -> Array a

13331333 -> Fold m a b

@@ -1348,13 +1348,14 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

13481348 if patLen == 0

13491349 then return $ Skip $ GO_EMPTY_PAT state

13501350 else if patLen == 1

1351− then return $ Skip $ GO_SINGLE_PAT state (unsafeIndex patArr 0)

1351+ then do

1352+ r <- liftIO $ (A.unsafeIndexIO patArr 0)

1353+ return $ Skip $ GO_SINGLE_PAT state r

13521354 else if sizeOf (undefined :: a) * patLen

13531355 <= sizeOf (undefined :: Word)

13541356 then return $ Skip $ GO_SHORT_PAT state

13551357 else do

1356− let !(rb, rhead) = unsafeDupablePerformIO $

1357− RB.unsafeNew patLen

1358+ (rb, rhead) <- liftIO $ RB.new patLen

13581359 return $ Skip $ GO_KARP_RABIN state rb rhead

13591360

13601361 stepOuter gst (GO_SINGLE_PAT stt pat) = do

@@ -1484,10 +1485,10 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

14841485 Yield x s -> do

14851486 acc <- initial

14861487 acc' <- if withSep then fstep acc x else return acc

1487− let !rh' = unsafeInlineIO (RB.insert rb rhead x)

1488+ rh' <- liftIO (RB.unsafeInsert rb rhead x)

14881489 if idx == maxIndex

14891490 then do

1490− let fold = RB.foldRing (RB.ringBound rb)

1491+ let fold = RB.unsafeFoldRing (RB.ringBound rb)

14911492 let !ringHash = fold addCksum 0 rb

14921493 if ringHash == patHash

14931494 then go2 SPEC ringHash rh' s acc'

@@ -1513,10 +1514,10 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

15131514 case res of

15141515 Yield x s -> do

15151516 acc' <- if withSep then fstep acc x else return acc

1516− let !rh' = unsafeInlineIO (RB.insert rb rh x)

1517+ rh' <- liftIO (RB.unsafeInsert rb rh x)

15171518 if idx == maxIndex

15181519 then do

1519− let fold = RB.foldRing (RB.ringBound rb)

1520+ let fold = RB.unsafeFoldRing (RB.ringBound rb)

15201521 let !ringHash = fold addCksum 0 rb

15211522 if ringHash == patHash

15221523 then go2 SPEC ringHash rh' s acc'

@@ -1529,7 +1530,7 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

15291530 then return Stop

15301531 else do

15311532 !acc' <- if idx /= 0 && not withSep

1532− then RB.foldRingM rh fstep acc rb

1533+ then RB.unsafeFoldRingM rh fstep acc rb

15331534 else return acc

15341535 done acc' >>= \r -> return $ Yield r GO_DONE

15351536

@@ -1543,18 +1544,18 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

15431544 res <- step (adaptState gst) st

15441545 case res of

15451546 Yield x s -> do

1546− let !old = unsafeInlineIO $ peek rh

1547− cksum' = deltaCksum cksum old x

1547+ old <- liftIO $ peek rh

1548+ let cksum' = deltaCksum cksum old x

15481549 acc' <- if withSep

15491550 then fstep acc x

15501551 else fstep acc old

15511552

15521553 if (cksum' == patHash)

15531554 then do

1554− let !rh'= unsafeInlineIO (RB.insert rb rh x)

1555+ rh' <- liftIO (RB.unsafeInsert rb rh x)

15551556 go2 SPEC cksum' rh' s acc'

15561557 else do

1557− let !rh'= unsafeInlineIO (RB.insert rb rh x)

1558+ rh' <- liftIO (RB.unsafeInsert rb rh x)

15581559 go1 SPEC cksum' rh' s acc'

15591560 Skip s -> go1 SPEC cksum rh s acc

15601561 Stop -> do

@@ -1563,7 +1564,7 @@ splitSuffixOn withSep patArr@Array{..} (Fold fstep initial done)

15631564 else do

15641565 acc' <- if withSep

15651566 then return acc

1566− else RB.foldRingFullM rh fstep acc rb

1567+ else RB.unsafeFoldRingFullM rh fstep acc rb

15671568 done acc' >>= \r -> return $ Yield r GO_DONE

15681569

15691570 go2 !_ !cksum' !rh' s !acc' = do

test/Arrays.hs+36−23

@@ -3,12 +3,12 @@

44module Main (main) where

6−import Data.Functor.Identity (runIdentity)

76import Foreign.Storable (Storable(..))

98import Test.Hspec.QuickCheck

10−import Test.QuickCheck (Property, forAll, Gen, vectorOf, arbitrary, (===),

9+import Test.QuickCheck (Property, forAll, Gen, vectorOf, arbitrary,

1110 choose)

11+import Test.QuickCheck.Monadic (monadicIO, assert)

1212

1313import Test.Hspec as H

1414

@@ -41,52 +41,64 @@ testLength :: Property

4141testLength =

4242 forAll (choose (0, maxArrLen)) $ \len ->

4343 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

44− let arr = runIdentity $ A.fromStreamN len

45− $ S.fromList list

46− in A.length arr === len

44+ monadicIO $ do

45+ arr <- A.fromStreamN len

46+ $ S.fromList list

47+ assert (A.length arr == len)

4748

4849testFromToStreamN :: Property

4950testFromToStreamN =

5051 forAll (choose (0, maxArrLen)) $ \len ->

5152 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

52− let arr = runIdentity $ A.fromStreamN len

53− $ S.fromList list

54− in runIdentity (S.toList (A.toStream arr)) === list

53+ monadicIO $ do

54+ arr <- A.fromStreamN len

55+ $ S.fromList list

56+ xs <- S.toList

57+ $ A.toStream arr

58+ assert (xs == list)

59+

60+testToStreamRev :: Property

61+testToStreamRev =

62+ forAll (choose (0, maxArrLen)) $ \len ->

63+ forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

64+ monadicIO $ do

65+ arr <- A.fromStreamN len

66+ $ S.fromList list

67+ xs <- S.toList

68+ $ A.toStreamRev arr

69+ assert (xs == reverse list)

5570

5671testArraysOf :: Property

5772testArraysOf =

5873 forAll (choose (0, maxArrLen)) $ \len ->

5974 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

60− let xs = runIdentity

61− $ S.toList

75+ monadicIO $ do

76+ xs <- S.toList

6277 $ S.concatMap A.toStream

6378 $ A.arraysOf 240

6479 $ S.fromList list

65− in xs === list

80+ assert (xs == list)

6681

6782testFlattenArrays :: Property

6883testFlattenArrays =

6984 forAll (choose (0, maxArrLen)) $ \len ->

7085 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

71− let xs = runIdentity

72− $ S.toList

86+ monadicIO $ do

87+ xs <- S.toList

7388 $ A.flattenArrays

7489 $ A.arraysOf 240

7590 $ S.fromList list

76− in xs === list

91+ assert (xs == list)

7792

78−{-

7993testFromToStream :: Property

8094testFromToStream =

8195 forAll (choose (0, maxArrLen)) $ \len ->

8296 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

83− let xs = runIdentity

84− $ S.toList

85− $ A.toStream

86− $ runIdentity. A.fromStream

87− $ S.fromList list

88− in xs === list

89−-}

97+ monadicIO $ do

98+ arr <- A.fromStream $ S.fromList list

99+ xs <- S.toList

100+ $ A.toStream arr

101+ assert (xs == list)

90102

91103main :: IO ()

92104main = hspec

@@ -96,6 +108,7 @@ main = hspec

96108 describe "Construction" $ do

97109 prop "length . fromStreamN n === n" $ testLength

98110 prop "toStream . fromStreamN n === id" $ testFromToStreamN

111+ prop "toStreamRev . fromStream === reverse" $ testToStreamRev

99112 prop "arraysOf concats to original" $ testArraysOf

100113 prop "flattenArrays concats to original" $ testFlattenArrays

101− -- prop "toStream . fromStream === id" $ testFromToStream

114+ prop "toStream . fromStream === id" $ testFromToStream

test/Prop.hs+1

@@ -452,6 +452,7 @@ transformCombineOpsCommon constr desc eq t = do

452452

453453 -- reordering

454454 prop (desc <> " reverse") $ transform reverse t S.reverse

455+ prop (desc <> " reverse'") $ transform reverse t S.reverse'

455456

456457 -- inserting

457458 prop (desc <> " intersperseM") $