git.iko.soy/streamly
on 3a90efb help sign in
streamly/test/Streamly/Test/Data/Stream/Async.hs
322 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)
19#if __GLASGOW_HASKELL__ < 808
20import Data.Semigroup ((<>))
21#endif
22import Data.Word (Word8)
23import Streamly.Data.Stream (Stream)
24import Test.Hspec.QuickCheck
25import Test.QuickCheck (Testable, Property, choose, forAll, withMaxSuccess)
26import Test.QuickCheck.Monadic (monadicIO, run)
27import Test.Hspec as H
29import qualified Streamly.Data.Fold as Fold
30import qualified Streamly.Data.Stream as Stream
31import qualified Streamly.Internal.Data.Stream.Async as Async
33import Streamly.Test.Common (listEquals)
35moduleName :: String
36moduleName = "Data.Stream.Async"
38-------------------------------------------------------------------------------
39-- Utilities
40-------------------------------------------------------------------------------
42sortEq :: Ord a => [a] -> [a] -> Bool
43sortEq a b = sort a == sort b
45cmp :: Stream IO Int -> ([Int] -> [Int] -> Bool) -> [Int] -> Property
46cmp s eq list =
47 monadicIO $ do
48 stream <- run $ Stream.fold Fold.toList s
49 listEquals eq stream list
51prop1 :: Testable prop => String -> prop -> SpecWith ()
52prop1 x y = modifyMaxSuccess (const 1) $ prop x y
54-- Coverage build takes too long with default number of tests
55maxTestCount :: Int
56#ifdef DEVBUILD
57maxTestCount = 100
58#else
59maxTestCount = 10
60#endif
62transformCombineFromList ::
63 ([Int] -> Stream IO Int)
64 -> ([Int] -> [Int] -> Bool)
65 -> ([Int] -> [Int])
66 -> (Stream IO Int -> Stream IO Int)
67 -> [Int]
68 -> [Int]
69 -> [Int]
70 -> Property
71transformCombineFromList constr eq listOp op a b c =
72 withMaxSuccess maxTestCount $
73 monadicIO $ do
74 stream <- run (Stream.fold Fold.toList $
75 constr a `Async.append` op (constr b `Async.append` constr c))
76 let list = a <> listOp (b <> c)
77 listEquals eq stream list
79commonOpConfigs :: [(String, Async.Config -> Async.Config)]
80commonOpConfigs =
81 [ ("default", id)
82#ifndef COVERAGE_BUILD
83 , ("rate AvgRate 10000", Async.avgRate 10000)
84 , ("rate Nothing", Async.rate Nothing)
85 , ("maxBuffer 0", Async.maxBuffer 0)
86 , ("maxThreads 0", Async.maxThreads 0)
87 , ("maxThreads 1", Async.maxThreads 1)
88#ifdef USE_LARGE_MEMORY
89 , ("maxThreads -1", Async.maxThreads (-1))
90#endif
91#endif
92 ]
94opConfigs :: [(String, Async.Config -> Async.Config)]
95opConfigs = commonOpConfigs
96 ++ [
97#ifndef COVERAGE_BUILD
98 ("maxBuffer 1", Async.maxBuffer 1)
99#endif
100 ]
102makeSpec :: [(String, a)] -> (a -> Spec) -> Spec
103makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg
104asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec
105asyncSpec =
106 makeSpec $ opConfigs
107#ifndef COVERAGE_BUILD
108 <> [("maxBuffer (-1)", Async.maxBuffer (-1))]
109#endif
111-------------------------------------------------------------------------------
112-- Compose with MonadThrow
113-------------------------------------------------------------------------------
115newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)
117instance Exception ExampleException
119exceptionPropagation ::
120 (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
121exceptionPropagation f = do
122 it "append throwM, nil" $
123 try (tl (throwM (ExampleException "E") `f` Stream.nil))
124 `shouldReturn`
125 (Left (ExampleException "E") :: Either ExampleException [Int])
126 it "append nil, throwM" $
127 try (tl (Stream.nil `f` throwM (ExampleException "E")))
128 `shouldReturn`
129 (Left (ExampleException "E") :: Either ExampleException [Int])
130 it "append nested throwM" $ do
131 let nested =
132 Stream.fromFoldable [1..10]
133 `f` throwM (ExampleException "E")
134 `f` Stream.fromFoldable [1..10]
135 try (tl (Stream.nil `f` nested `f` Stream.fromFoldable [1..10]))
136 `shouldReturn`
137 (Left (ExampleException "E")
138 :: Either ExampleException [Int])
139 it "sequence throwM" $
140 let stream = Stream.fromList [throwM (ExampleException "E")]
141 in try (tl (Stream.nil `f` Async.sequence stream))
142 `shouldReturn`
143 (Left (ExampleException "E") :: Either ExampleException [Int])
145 it "concatMap throwM" $ do
146 let s1 = Async.concatList $ fmap Stream.fromPure [1..4]
147 s2 = Async.concatList $ fmap Stream.fromPure [5..8]
148 try $ tl (
149 let bind = flip Async.concatMap
150 in bind s1 $ \x ->
151 bind s2 $ \y ->
152 if x + y > 10
153 then throwM (ExampleException "E")
154 else return (x + y)
155 )
156 `shouldReturn`
157 (Left (ExampleException "E") :: Either ExampleException [Int])
159 where
161 tl = Stream.fold Fold.toList
163---------------------------------------------------------------------------
164-- Time ordering
165---------------------------------------------------------------------------
167#ifdef DEVBUILD
168timeOrdering :: (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
169timeOrdering f = do
170 it "Parallel ordering left associated" $
171 Stream.fold Fold.toList (((event 4 `f` event 3) `f` event 2) `f` event 1)
172 `shouldReturn` [1..4]
174 it "Parallel ordering right associated" $
175 Stream.fold Fold.toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
176 `shouldReturn` [1..4]
178 where event n = Stream.fromEffect (threadDelay (n * 200000)) >> return n
179#endif
181-------------------------------------------------------------------------------
182-- Some ad-hoc tests that failed at times
183-------------------------------------------------------------------------------
185takeCombined :: Int -> IO ()
186takeCombined n = do
187 let constr = Stream.fromFoldable
188 r <- Stream.fold Fold.toList $
189 Stream.take n
190 (constr ([] :: [Int]) `Async.append` constr ([] :: [Int]))
191 r `shouldBe` []
193---------------------------------------------------------------------------
194-- Main
195---------------------------------------------------------------------------
197constructWithLenM
198 :: (Int -> Stream IO Int)
199 -> (Int -> IO [Int])
200 -> Word8
201 -> Property
202constructWithLenM mkStream mkList len =
203 withMaxSuccess maxTestCount
204 $ monadicIO $ do
205 stream <-
206 run
207 $ Stream.fold Fold.toList
208 $ mkStream (fromIntegral len)
209 list <- run $ mkList (fromIntegral len)
210 listEquals (==) stream list
212sequenceReplicate
213 :: (Async.Config -> Async.Config)
214 -> Word8
215 -> Property
216sequenceReplicate cfg = constructWithLenM stream list
218 where
220 list = flip replicateM (return 1 :: IO Int)
221 stream = Async.sequenceWith cfg . flip Stream.replicate (return 1 :: IO Int)
223main :: IO ()
224main = hspec
225 $ H.parallel
226#ifdef COVERAGE_BUILD
227 $ modifyMaxSuccess (const 10)
228#endif
229 $ describe moduleName $ do
230 let transform = transformCombineFromList Stream.fromList sortEq
232 prop "eval" $
233 transform
234 (fmap (+2))
235 (fmap (+1) . Async.eval . fmap (+1))
237 asyncSpec $ prop "sequenceWith" . sequenceReplicate
238 -- XXX Need to use asyncSpec in all tests
239 prop "mapM (+1)" $
240 transform (fmap (+1)) (Async.mapM (\x -> return (x + 1)))
242 -- Binary append
243 prop1 "append [] []"
244 $ cmp (Stream.nil `Async.append` Stream.nil) sortEq []
245 prop1 "append [] [1]"
246 $ cmp (Stream.nil `Async.append` Stream.fromPure 1) sortEq [1]
247 prop1 "append [1] []"
248 $ cmp (Stream.fromPure 1 `Async.append` Stream.nil) sortEq [1]
249 prop1 "append [0] [1]"
250 $ let stream =
251 Stream.fromPure 0 `Async.append` Stream.fromPure 1
252 in cmp stream sortEq [0, 1]
254 prop1 "append [0] [] [1]"
255 $ let stream =
256 Stream.fromPure 0
257 `Async.append` Stream.nil
258 `Async.append` Stream.fromPure 1
259 in cmp stream sortEq [0, 1]
261 prop1 "append left associated"
262 $ let stream =
263 Stream.fromPure 0
264 `Async.append` Stream.fromPure 1
265 `Async.append` Stream.fromPure 2
266 `Async.append` Stream.fromPure 3
267 in cmp stream sortEq [0, 1, 2, 3]
269 prop1 "append right associated"
270 $ let stream =
271 Stream.fromPure 0
272 `Async.append` (Stream.fromPure 1
273 `Async.append` (Stream.fromPure 2
274 `Async.append` Stream.fromPure 3))
275 in cmp stream sortEq [0, 1, 2, 3]
277 prop1 "append balanced"
278 $ let leaf x y = Stream.fromPure x `Async.append` Stream.fromPure y
279 leaf11 = leaf 0 1 `Async.append` leaf 2 (3 :: Int)
280 leaf12 = leaf 4 5 `Async.append` leaf 6 7
281 stream = leaf11 `Async.append` leaf12
282 in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
284 prop1 "appendWith (maxThreads 1)"
285 $ let stream =
286 Async.appendWith (Async.maxThreads 1)
287 (Stream.fromList [1,2,3,4,5])
288 (Stream.fromList [6,7,8,9,10])
289 in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]
291 prop1 "apply (async arg1)"
292 $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append` pure 2)
293 s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)
294 xs = Stream.fold Fold.toList s2
295 in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]
297 prop1 "apply (async arg2)"
298 $ let s1 = pure (1,)
299 s2 = Async.apply s1 (pure 2 `Async.append` pure 3)
300 xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]
301 in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]
303 -- concat
304 prop1 "concat"
305 $ let stream =
306 Async.concat
307 $ fmap Stream.fromPure
308 $ Stream.fromList [1..100]
309 in cmp stream sortEq [1..100]
311 prop "concatMap" $
312 forAll (choose (0, 100)) $ \n ->
313 transform
314 (concatMap (const [1..n]))
315 (Async.concatMap (const (Stream.fromList [1..n])))
317#ifdef DEVBUILD
318 describe "Time ordering" $ timeOrdering Async.append
319#endif
320 describe "Exception propagation" $ exceptionPropagation Async.append
321 -- Ad-hoc tests
322 it "takes n from stream of streams" $ takeCombined 2