git.iko.soy/streamly
on 4eaeb10 help sign in
streamly/test/Streamly/Test/Data/Stream/Async.hs
319 lines 11 kB history raw
1-- |
2-- Module : Streamly.Test.Data.Stream.Async
3-- Copyright : (c) 2020 Composewell Technologies
4--
5-- License : BSD-3-Clause
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
8-- Portability : GHC
10module Streamly.Test.Data.Stream.Async (main) where
12#ifdef DEVBUILD
13import Control.Concurrent (threadDelay)
14#endif
15import Control.Exception (Exception, try)
16import Control.Monad (replicateM)
17import Control.Monad.Catch (throwM)
18import Data.List (sort)
19import Data.Word (Word8)
20import Streamly.Data.Stream (Stream)
21import Test.Hspec.QuickCheck
22import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)
23import Test.QuickCheck.Monadic (monadicIO, run)
24import Test.Hspec as H
26import qualified Streamly.Data.Fold as Fold
27import qualified Streamly.Data.Stream as Stream
28import qualified Streamly.Internal.Data.Stream.Async as Async
30import Streamly.Test.Common (listEquals)
32moduleName :: String
33moduleName = "Data.Stream.Async"
35-------------------------------------------------------------------------------
36-- Utilities
37-------------------------------------------------------------------------------
39sortEq :: Ord a => [a] -> [a] -> Bool
40sortEq a b = sort a == sort b
42cmp :: Stream IO Int -> ([Int] -> [Int] -> Bool) -> [Int] -> Property
43cmp s eq list =
44 monadicIO $ do
45 stream <- run $ Stream.fold Fold.toList s
46 listEquals eq stream list
48prop1 :: Testable prop => String -> prop -> SpecWith ()
49prop1 x y = modifyMaxSuccess (const 1) $ prop x y
51-- Coverage build takes too long with default number of tests
52maxTestCount :: Int
53#ifdef DEVBUILD
54maxTestCount = 100
55#else
56maxTestCount = 10
57#endif
59transformCombineFromList ::
60 ([Int] -> Stream IO Int)
61 -> ([Int] -> [Int] -> Bool)
62 -> ([Int] -> [Int])
63 -> (Stream IO Int -> Stream IO Int)
64 -> [Int]
65 -> [Int]
66 -> [Int]
67 -> Property
68transformCombineFromList constr eq listOp op a b c =
69 withMaxSuccess maxTestCount $
70 monadicIO $ do
71 stream <- run (Stream.fold Fold.toList $
72 constr a `Async.append` op (constr b `Async.append` constr c))
73 let list = a <> listOp (b <> c)
74 listEquals eq stream list
76commonOpConfigs :: [(String, Async.Config -> Async.Config)]
77commonOpConfigs =
78 [ ("default", id)
79#ifndef COVERAGE_BUILD
80 , ("rate AvgRate 10000", Async.avgRate 10000)
81 , ("rate Nothing", Async.rate Nothing)
82 , ("maxBuffer 0", Async.maxBuffer 0)
83 , ("maxThreads 0", Async.maxThreads 0)
84 , ("maxThreads 1", Async.maxThreads 1)
85#ifdef USE_LARGE_MEMORY
86 , ("maxThreads -1", Async.maxThreads (-1))
87#endif
88#endif
89 ]
91opConfigs :: [(String, Async.Config -> Async.Config)]
92opConfigs = commonOpConfigs
93 ++ [
94#ifndef COVERAGE_BUILD
95 ("maxBuffer 1", Async.maxBuffer 1)
96#endif
97 ]
99makeSpec :: [(String, a)] -> (a -> Spec) -> Spec
100makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg
101asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec
102asyncSpec =
103 makeSpec $ opConfigs
104#ifndef COVERAGE_BUILD
105 <> [("maxBuffer (-1)", Async.maxBuffer (-1))]
106#endif
108-------------------------------------------------------------------------------
109-- Compose with MonadThrow
110-------------------------------------------------------------------------------
112newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)
114instance Exception ExampleException
116exceptionPropagation ::
117 (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
118exceptionPropagation f = do
119 it "append throwM, nil" $
120 try (tl (throwM (ExampleException "E") `f` Stream.nil))
121 `shouldReturn`
122 (Left (ExampleException "E") :: Either ExampleException [Int])
123 it "append nil, throwM" $
124 try (tl (Stream.nil `f` throwM (ExampleException "E")))
125 `shouldReturn`
126 (Left (ExampleException "E") :: Either ExampleException [Int])
127 it "append nested throwM" $ do
128 let nested =
129 Stream.fromFoldable [1..10]
130 `f` throwM (ExampleException "E")
131 `f` Stream.fromFoldable [1..10]
132 try (tl (Stream.nil `f` nested `f` Stream.fromFoldable [1..10]))
133 `shouldReturn`
134 (Left (ExampleException "E")
135 :: Either ExampleException [Int])
136 it "sequence throwM" $
137 let stream = Stream.fromList [throwM (ExampleException "E")]
138 in try (tl (Stream.nil `f` Async.sequence stream))
139 `shouldReturn`
140 (Left (ExampleException "E") :: Either ExampleException [Int])
142 it "concatMap throwM" $ do
143 let s1 = Async.concatList $ fmap Stream.fromPure [1..4]
144 s2 = Async.concatList $ fmap Stream.fromPure [5..8]
145 try $ tl (
146 let bind = flip Async.concatMap
147 in bind s1 $ \x ->
148 bind s2 $ \y ->
149 if x + y > 10
150 then throwM (ExampleException "E")
151 else return (x + y)
152 )
153 `shouldReturn`
154 (Left (ExampleException "E") :: Either ExampleException [Int])
156 where
158 tl = Stream.fold Fold.toList
160---------------------------------------------------------------------------
161-- Time ordering
162---------------------------------------------------------------------------
164#ifdef DEVBUILD
165timeOrdering :: (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
166timeOrdering f = do
167 it "Parallel ordering left associated" $
168 Stream.fold Fold.toList (((event 4 `f` event 3) `f` event 2) `f` event 1)
169 `shouldReturn` [1..4]
171 it "Parallel ordering right associated" $
172 Stream.fold Fold.toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
173 `shouldReturn` [1..4]
175 where event n = Stream.fromEffect (threadDelay (n * 200000)) >> return n
176#endif
178-------------------------------------------------------------------------------
179-- Some ad-hoc tests that failed at times
180-------------------------------------------------------------------------------
182takeCombined :: Int -> IO ()
183takeCombined n = do
184 let constr = Stream.fromFoldable
185 r <- Stream.fold Fold.toList $
186 Stream.take n
187 (constr ([] :: [Int]) `Async.append` constr ([] :: [Int]))
188 r `shouldBe` []
190---------------------------------------------------------------------------
191-- Main
192---------------------------------------------------------------------------
194constructWithLenM
195 :: (Int -> Stream IO Int)
196 -> (Int -> IO [Int])
197 -> Word8
198 -> Property
199constructWithLenM mkStream mkList len =
200 withMaxSuccess maxTestCount
201 $ monadicIO $ do
202 stream <-
203 run
204 $ Stream.fold Fold.toList
205 $ mkStream (fromIntegral len)
206 list <- run $ mkList (fromIntegral len)
207 listEquals (==) stream list
209sequenceReplicate
210 :: (Async.Config -> Async.Config)
211 -> Word8
212 -> Property
213sequenceReplicate cfg = constructWithLenM stream list
215 where
217 list = flip replicateM (return 1 :: IO Int)
218 stream = Async.sequenceWith cfg . flip Stream.replicate (return 1 :: IO Int)
220main :: IO ()
221main = hspec
222 $ H.parallel
223#ifdef COVERAGE_BUILD
224 $ modifyMaxSuccess (const 10)
225#endif
226 $ describe moduleName $ do
227 let transform = transformCombineFromList Stream.fromList sortEq
229 prop "eval" $
230 transform
231 (fmap (+2))
232 (fmap (+1) . Async.eval . fmap (+1))
234 asyncSpec $ prop "sequenceWith" . sequenceReplicate
235 -- XXX Need to use asyncSpec in all tests
236 prop "mapM (+1)" $
237 transform (fmap (+1)) (Async.mapM (\x -> return (x + 1)))
239 -- Binary append
240 prop1 "append [] []"
241 $ cmp (Stream.nil `Async.append` Stream.nil) sortEq []
242 prop1 "append [] [1]"
243 $ cmp (Stream.nil `Async.append` Stream.fromPure 1) sortEq [1]
244 prop1 "append [1] []"
245 $ cmp (Stream.fromPure 1 `Async.append` Stream.nil) sortEq [1]
246 prop1 "append [0] [1]"
247 $ let stream =
248 Stream.fromPure 0 `Async.append` Stream.fromPure 1
249 in cmp stream sortEq [0, 1]
251 prop1 "append [0] [] [1]"
252 $ let stream =
253 Stream.fromPure 0
254 `Async.append` Stream.nil
255 `Async.append` Stream.fromPure 1
256 in cmp stream sortEq [0, 1]
258 prop1 "append left associated"
259 $ let stream =
260 Stream.fromPure 0
261 `Async.append` Stream.fromPure 1
262 `Async.append` Stream.fromPure 2
263 `Async.append` Stream.fromPure 3
264 in cmp stream sortEq [0, 1, 2, 3]
266 prop1 "append right associated"
267 $ let stream =
268 Stream.fromPure 0
269 `Async.append` (Stream.fromPure 1
270 `Async.append` (Stream.fromPure 2
271 `Async.append` Stream.fromPure 3))
272 in cmp stream sortEq [0, 1, 2, 3]
274 prop1 "append balanced"
275 $ let leaf x y = Stream.fromPure x `Async.append` Stream.fromPure y
276 leaf11 = leaf 0 1 `Async.append` leaf 2 (3 :: Int)
277 leaf12 = leaf 4 5 `Async.append` leaf 6 7
278 stream = leaf11 `Async.append` leaf12
279 in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
281 prop1 "appendWith (maxThreads 1)"
282 $ let stream =
283 Async.appendWith (Async.maxThreads 1)
284 (Stream.fromList [1,2,3,4,5])
285 (Stream.fromList [6,7,8,9,10])
286 in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]
288 prop1 "apply (async arg1)"
289 $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append` pure 2)
290 s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)
291 xs = Stream.fold Fold.toList s2
292 in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]
294 prop1 "apply (async arg2)"
295 $ let s1 = pure (1,)
296 s2 = Async.apply s1 (pure 2 `Async.append` pure 3)
297 xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]
298 in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]
300 -- concat
301 prop1 "concat"
302 $ let stream =
303 Async.concat
304 $ fmap Stream.fromPure
305 $ Stream.fromList [1..100]
306 in cmp stream sortEq [1..100]
308 prop "concatMap" $
309 forAll (choose (0, 100)) $ \n ->
310 transform
311 (concatMap (const [1..n]))
312 (Async.concatMap (const (Stream.fromList [1..n])))
314#ifdef DEVBUILD
315 describe "Time ordering" $ timeOrdering Async.append
316#endif
317 describe "Exception propagation" $ exceptionPropagation Async.append
318 -- Ad-hoc tests
319 it "takes n from stream of streams" $ takeCombined 2