cff0369 · streamly · git.iko.soy

git.iko.soy/streamly

Refactor ops from Prelude to Prelude.Serial.

commitcff03690c67dece05402109ed3fc4f94a637c38f authorpranaysashank <pranaysashank@composewell.com> dateWed 14 Oct 2020, 14:28 (6 years ago) parenta16a716

2 files changed+112−115

compare with master

unifiedsplit

test/Streamly/Test/Prelude.hs+1−114 test/Streamly/Test/Prelude/Serial.hs+111−1

test/Streamly/Test/Prelude.hs+1−114

@@ -11,127 +11,24 @@ module Streamly.Test.Prelude (main) where

1111

1212import Control.Concurrent (myThreadId, threadDelay)

1313import Control.Exception (Exception, try)

14−import Control.Monad (when)

1514import Control.Monad.Catch (throwM)

1615import Control.Monad.Error.Class (throwError, MonadError)

1716import Control.Monad.Trans.Except (runExceptT, ExceptT)

18−import Data.Function ((&))

19−import Data.IORef (newIORef, readIORef, writeIORef)

20−import Data.Int (Int64)

2117import Data.List (sort)

22−import Data.Maybe (fromJust, isJust)

2318#if !(MIN_VERSION_base(4,11,0))

2419import Data.Semigroup (Semigroup(..))

2520#endif

2621import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

2722import System.Random (randomIO)

2823import Test.Hspec as H

29−import Test.Hspec.QuickCheck (prop)

30−import Test.QuickCheck (Property, choose)

31−import Test.QuickCheck.Monadic (monadicIO, pick)

3224

3325import Streamly.Prelude (SerialT, IsStream)

34−import Streamly.Internal.Data.Time.Units

35− (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)

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

3726

38−import qualified Streamly.Internal.Data.Fold as FL

39−import qualified Streamly.Internal.Data.Unfold as UF

40−import qualified Streamly.Internal.Data.Stream.IsStream as IS

4127import qualified Streamly.Prelude as S

4228

4329toListSerial :: SerialT IO a -> IO [a]

4430toListSerial = S.toList . S.serially

4531

46−max_length :: Int

47−max_length = 1000

48−

49−tenPow8 :: Int64

50−tenPow8 = 10^(8 :: Int)

51−

52−tenPow7 :: Int64

53−tenPow7 = 10^(7 :: Int)

54−

55−takeDropTime :: NanoSecond64

56−takeDropTime = NanoSecond64 $ 5 * tenPow8

57−

58−checkTakeDropTime :: (Maybe AbsTime, Maybe AbsTime) -> IO Bool

59−checkTakeDropTime (mt0, mt1) = do

60− let graceTime = NanoSecond64 $ 8 * tenPow7

61− case mt0 of

62− Nothing -> return True

63− Just t0 ->

64− case mt1 of

65− Nothing -> return True

66− Just t1 -> do

67− let tMax = toRelTime64 (takeDropTime + graceTime)

68− let tMin = toRelTime64 (takeDropTime - graceTime)

69− let t = diffAbsTime64 t1 t0

70− let r = t >= tMin && t <= tMax

71− when (not r) $ putStrLn $

72− "t = " ++ show t ++

73− " tMin = " ++ show tMin ++

74− " tMax = " ++ show tMax

75− return r

76−

77−testTakeByTime :: IO Bool

78−testTakeByTime = do

79− r <-

80− S.fold ((,) <$> FL.head <*> FL.last)

81− $ IS.takeByTime takeDropTime

82− $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

83− checkTakeDropTime r

84−

85−testDropByTime :: IO Bool

86−testDropByTime = do

87− t0 <- getTime Monotonic

88− mt1 <-

89− S.head

90− $ IS.dropByTime takeDropTime

91− $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

92− checkTakeDropTime (Just t0, mt1)

93−

94−unfold :: Property

95−unfold = monadicIO $ do

96− a <- pick $ choose (0, max_length `div` 2)

97− b <- pick $ choose (0, max_length)

98− let unf = UF.enumerateFromToIntegral b

99− ls <- S.toList $ S.unfold unf a

100− return $ ls == [a..b]

101−

102−unfold0 :: Property

103−unfold0 = monadicIO $ do

104− a <- pick $ choose (0, max_length `div` 2)

105− b <- pick $ choose (0, max_length)

106− let unf = UF.supply (UF.enumerateFromToIntegral b) a

107− ls <- S.toList $ IS.unfold0 unf

108− return $ ls == [a..b]

109−

110−testFromCallback :: IO Int

111−testFromCallback = do

112− ref <- newIORef Nothing

113− let stream = S.map Just (IS.fromCallback (setCallback ref))

114− `S.parallel` runCallback ref

115− S.sum $ S.map fromJust $ S.takeWhile isJust stream

116−

117− where

118−

119− setCallback ref cb = do

120− writeIORef ref (Just cb)

121−

122− runCallback ref = S.yieldM $ do

123− cb <-

124− S.repeatM (readIORef ref)

125− & IS.delayPost 0.1

126− & S.mapMaybe id

127− & S.head

128−

129− S.fromList [1..100]

130− & IS.delayPost 0.001

131− & S.mapM_ (fromJust cb)

132− threadDelay 100000

133− return Nothing

134−

13532-- XXX need to test that we have promptly cleaned up everything after the error

13633-- XXX We can also check the output that we are expected to get before the

13734-- error occurs.

@@ -275,16 +172,6 @@ parallelLoops = do

275172

276173main :: IO ()

277174main = hspec $ H.parallel $ do

278− describe "Filtering" $ do

279− it "takeByTime" (testTakeByTime `shouldReturn` True)

280− it "dropByTime" (testDropByTime `shouldReturn` True)

281−

282− describe "From Generators" $ do

283− prop "unfold" unfold

284− prop "unfold0" unfold0

285−

286− it "fromCallback" $ testFromCallback `shouldReturn` (50*101)

287−

288175 describe "Miscellaneous combined examples" mixedOps

289176 describe "Miscellaneous combined examples aheadly" mixedOpsAheadly

290177 describe "Simple MonadError and MonadThrow" simpleMonadError

@@ -296,4 +183,4 @@ main = hspec $ H.parallel $ do

296183 describe "Composed MonadError wSerially" $ composeWithMonadError wSerially

297184 describe "Composed MonadError asyncly" $ composeWithMonadError asyncly

298185 describe "Composed MonadError wAsyncly" $ composeWithMonadError wAsyncly

299− -}

186+ -}

test/Streamly/Test/Prelude/Serial.hs+111−1

@@ -9,7 +9,13 @@

1010module Streamly.Test.Prelude.Serial where

1111

12+import Control.Concurrent ( threadDelay )

13+import Control.Monad ( when )

14+import Data.Function ( (&) )

15+import Data.IORef ( newIORef, readIORef, writeIORef )

16+import Data.Int (Int64)

1217import Data.List (intercalate)

18+import Data.Maybe ( isJust, fromJust )

1319#if __GLASGOW_HASKELL__ < 808

1420import Data.Semigroup ((<>))

1521#endif

@@ -26,13 +32,18 @@ import Test.QuickCheck

2632 , vectorOf

2733 , withMaxSuccess

2834 )

29−import Test.QuickCheck.Monadic (assert, monadicIO, run)

35+import Test.QuickCheck.Monadic (assert, monadicIO, pick, run)

3036import Test.Hspec as H

3137

3238import Streamly.Prelude

3339 ( SerialT, IsStream, avgRate, maxBuffer, serial, serially)

3440import qualified Streamly.Prelude as S

3541import qualified Streamly.Data.Fold as FL

42+import qualified Streamly.Internal.Data.Unfold as UF

43+import qualified Streamly.Internal.Data.Stream.IsStream as IS

44+import Streamly.Internal.Data.Time.Units

45+ (AbsTime, NanoSecond64(..), toRelTime64, diffAbsTime64)

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

3647

3748import Streamly.Test.Common

3849import Streamly.Test.Prelude.Common

@@ -163,6 +174,95 @@ associativityCheck desc t = prop desc assocCheckProp

163174 assocStream <- run $ S.toList $ t $ xStream <> yStream <> zStream

164175 listEquals (==) infixAssocstream assocStream

165176

177+max_length :: Int

178+max_length = 1000

179+

180+tenPow8 :: Int64

181+tenPow8 = 10^(8 :: Int)

182+

183+tenPow7 :: Int64

184+tenPow7 = 10^(7 :: Int)

185+

186+takeDropTime :: NanoSecond64

187+takeDropTime = NanoSecond64 $ 5 * tenPow8

188+

189+checkTakeDropTime :: (Maybe AbsTime, Maybe AbsTime) -> IO Bool

190+checkTakeDropTime (mt0, mt1) = do

191+ let graceTime = NanoSecond64 $ 8 * tenPow7

192+ case mt0 of

193+ Nothing -> return True

194+ Just t0 ->

195+ case mt1 of

196+ Nothing -> return True

197+ Just t1 -> do

198+ let tMax = toRelTime64 (takeDropTime + graceTime)

199+ let tMin = toRelTime64 (takeDropTime - graceTime)

200+ let t = diffAbsTime64 t1 t0

201+ let r = t >= tMin && t <= tMax

202+ when (not r) $ putStrLn $

203+ "t = " ++ show t ++

204+ " tMin = " ++ show tMin ++

205+ " tMax = " ++ show tMax

206+ return r

207+

208+testTakeByTime :: IO Bool

209+testTakeByTime = do

210+ r <-

211+ S.fold ((,) <$> FL.head <*> FL.last)

212+ $ IS.takeByTime takeDropTime

213+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

214+ checkTakeDropTime r

215+

216+testDropByTime :: IO Bool

217+testDropByTime = do

218+ t0 <- getTime Monotonic

219+ mt1 <-

220+ S.head

221+ $ IS.dropByTime takeDropTime

222+ $ S.repeatM (threadDelay 1000 >> getTime Monotonic)

223+ checkTakeDropTime (Just t0, mt1)

224+

225+unfold :: Property

226+unfold = monadicIO $ do

227+ a <- pick $ choose (0, max_length `div` 2)

228+ b <- pick $ choose (0, max_length)

229+ let unf = UF.enumerateFromToIntegral b

230+ ls <- S.toList $ S.unfold unf a

231+ return $ ls == [a..b]

232+

233+unfold0 :: Property

234+unfold0 = monadicIO $ do

235+ a <- pick $ choose (0, max_length `div` 2)

236+ b <- pick $ choose (0, max_length)

237+ let unf = UF.supply (UF.enumerateFromToIntegral b) a

238+ ls <- S.toList $ IS.unfold0 unf

239+ return $ ls == [a..b]

240+

241+testFromCallback :: IO Int

242+testFromCallback = do

243+ ref <- newIORef Nothing

244+ let stream = S.map Just (IS.fromCallback (setCallback ref))

245+ `S.parallel` runCallback ref

246+ S.sum $ S.map fromJust $ S.takeWhile isJust stream

247+

248+ where

249+

250+ setCallback ref cb = do

251+ writeIORef ref (Just cb)

252+

253+ runCallback ref = S.yieldM $ do

254+ cb <-

255+ S.repeatM (readIORef ref)

256+ & IS.delayPost 0.1

257+ & S.mapMaybe id

258+ & S.head

259+

260+ S.fromList [1..100]

261+ & IS.delayPost 0.001

262+ & S.mapM_ (fromJust cb)

263+ threadDelay 100000

264+ return Nothing

265+

166266main :: IO ()

167267main = hspec

168268 $ H.parallel

@@ -209,6 +309,10 @@ main = hspec

209309 serialOps $ prop "serially fromIndices" . constructWithFromIndices

210310 serialOps $ prop "serially fromIndicesM" . constructWithFromIndicesM

211311

312+ describe "From Generators" $ do

313+ prop "unfold" unfold

314+ prop "unfold0" unfold0

315+

212316 describe "Functor operations" $ do

213317 serialOps $ functorOps S.fromFoldable "serially" (==)

214318 serialOps $ functorOps folded "serially folded" (==)

@@ -269,6 +373,10 @@ main = hspec

269373 serialOps $ transformCombineOpsOrdered S.fromFoldable "serially" (==)

270374 serialOps $ transformCombineOpsOrdered folded "serially" (==)

271375

376+ describe "Filtering" $ do

377+ it "takeByTime" (testTakeByTime `shouldReturn` True)

378+ it "dropByTime" (testDropByTime `shouldReturn` True)

379+

272380 describe "Stream group and split operations" $ do

273381 groupSplitOps "serially"

274382

@@ -289,3 +397,5 @@ main = hspec

289397 prop "testGroupsBySep" testGroupsBySep

290398

291399 describe "Composed MonadThrow serially" $ composeWithMonadThrow serially

400+

401+ it "fromCallback" $ testFromCallback `shouldReturn` (50*101)