git.iko.soy/streamly
on ffcb268 help sign in
streamly/test/Streamly/Test/Data/Stream/Concurrent.hs
322 lines 11 kB history raw
1-- |
2-- Module : Streamly.Test.Data.Stream.Concurrent
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.Concurrent (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.Concurrent as Async
30import Streamly.Test.Common (listEquals)
32moduleName :: String
33moduleName = "Data.Stream.Concurrent"
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 let s1 = op (Async.parLazy [constr b, constr c])
72 let s2 = Async.parLazy [constr a, s1]
73 stream <- run (Stream.fold Fold.toList s2)
74 let list = a <> listOp (b <> c)
75 listEquals eq stream list
77commonOpConfigs :: [(String, Async.Config -> Async.Config)]
78commonOpConfigs =
79 [ ("default", id)
80#ifndef COVERAGE_BUILD
81 , ("rate AvgRate 10000", Async.avgRate 10000)
82 , ("rate Nothing", Async.rate Nothing)
83 , ("maxBuffer 0", Async.maxBuffer 0)
84 , ("maxThreads 0", Async.maxThreads 0)
85 , ("maxThreads 1", Async.maxThreads 1)
86 , ("eager", Async.eager True)
87 -- XXX Need to use an unsorted eq operation for ahead
88 , ("ordered", Async.ordered True)
89#ifdef USE_LARGE_MEMORY
90 , ("maxThreads -1", Async.maxThreads (-1))
91#endif
92#endif
93 ]
95opConfigs :: [(String, Async.Config -> Async.Config)]
96opConfigs = commonOpConfigs
97 ++ [
98#ifndef COVERAGE_BUILD
99 ("maxBuffer 1", Async.maxBuffer 1)
100#endif
101 ]
103makeSpec :: [(String, a)] -> (a -> Spec) -> Spec
104makeSpec cfg spec = mapM_ (\(desc, arg) -> describe desc $ spec arg) cfg
106asyncSpec :: ((Async.Config -> Async.Config) -> Spec) -> Spec
107asyncSpec =
108 makeSpec $ opConfigs
109#ifndef COVERAGE_BUILD
110 <> [("maxBuffer (-1)", Async.maxBuffer (-1))]
111#endif
113-------------------------------------------------------------------------------
114-- Compose with MonadThrow
115-------------------------------------------------------------------------------
117newtype ExampleException = ExampleException String deriving (Eq, Show, Ord)
119instance Exception ExampleException
121exceptionPropagation ::
122 (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
123exceptionPropagation f = do
124 it "append throwM, nil" $
125 try (tl (Stream.fromEffect (throwM (ExampleException "E")) `f` Stream.nil))
126 `shouldReturn`
127 (Left (ExampleException "E") :: Either ExampleException [Int])
128 it "append nil, throwM" $
129 try (tl (Stream.nil `f` Stream.fromEffect (throwM (ExampleException "E"))))
130 `shouldReturn`
131 (Left (ExampleException "E") :: Either ExampleException [Int])
132 it "append nested throwM" $ do
133 let nested =
134 Stream.fromFoldable [1..10]
135 `f` Stream.fromEffect (throwM (ExampleException "E"))
136 `f` Stream.fromFoldable [1..10]
137 try (tl (Stream.nil `f` nested `f` Stream.fromFoldable [1..10]))
138 `shouldReturn`
139 (Left (ExampleException "E")
140 :: Either ExampleException [Int])
141 it "sequence throwM" $
142 let stream = Stream.fromList [throwM (ExampleException "E")]
143 in try (tl (Stream.nil `f` Async.sequence stream))
144 `shouldReturn`
145 (Left (ExampleException "E") :: Either ExampleException [Int])
147 it "concatMap throwM" $ do
148 let s1 = Async.parConcatList id $ fmap Stream.fromPure [1..4]
149 s2 = Async.parConcatList id $ fmap Stream.fromPure [5..8]
150 try $ tl (
151 let bind = flip Async.concatMap
152 in bind s1 $ \x ->
153 bind s2 $ \y ->
154 if x + y > 10
155 then Stream.fromEffect (throwM (ExampleException "E"))
156 else Stream.fromPure (x + y)
157 )
158 `shouldReturn`
159 (Left (ExampleException "E") :: Either ExampleException [Int])
161 where
163 tl = Stream.fold Fold.toList
165---------------------------------------------------------------------------
166-- Time ordering
167---------------------------------------------------------------------------
169#ifdef DEVBUILD
170timeOrdering :: (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec
171timeOrdering f = do
172 it "Parallel ordering left associated" $
173 Stream.fold Fold.toList (((event 4 `f` event 3) `f` event 2) `f` event 1)
174 `shouldReturn` [1..4]
176 it "Parallel ordering right associated" $
177 Stream.fold Fold.toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))
178 `shouldReturn` [1..4]
180 where event n = Stream.fromEffect (threadDelay (n * 200000)) >> return n
181#endif
183-------------------------------------------------------------------------------
184-- Some ad-hoc tests that failed at times
185-------------------------------------------------------------------------------
187takeCombined :: Int -> IO ()
188takeCombined n = do
189 let constr = Stream.fromFoldable
190 let s = Async.parLazy [constr ([] :: [Int]), constr ([] :: [Int])]
191 r <- Stream.fold Fold.toList $ Stream.take n s
192 r `shouldBe` []
194---------------------------------------------------------------------------
195-- Main
196---------------------------------------------------------------------------
198constructWithLenM
199 :: (Int -> Stream IO Int)
200 -> (Int -> IO [Int])
201 -> Word8
202 -> Property
203constructWithLenM mkStream mkList len =
204 withMaxSuccess maxTestCount
205 $ monadicIO $ do
206 stream <-
207 run
208 $ Stream.fold Fold.toList
209 $ mkStream (fromIntegral len)
210 list <- run $ mkList (fromIntegral len)
211 listEquals (==) stream list
213sequenceReplicate
214 :: (Async.Config -> Async.Config)
215 -> Word8
216 -> Property
217sequenceReplicate cfg = constructWithLenM stream list
219 where
221 list = flip replicateM (return 1 :: IO Int)
222 stream = Async.parSequence cfg . flip Stream.replicate (return 1 :: IO Int)
224main :: IO ()
225main = hspec
226 $ H.parallel
227#ifdef COVERAGE_BUILD
228 $ modifyMaxSuccess (const 10)
229#endif
230 $ describe moduleName $ do
231 let transform = transformCombineFromList Stream.fromList sortEq
233 prop "eval" $
234 transform
235 (fmap (+2))
236 (fmap (+1) . Async.eval . fmap (+1))
238 asyncSpec $ prop "parSequence" . sequenceReplicate
239 -- XXX Need to use asyncSpec in all tests
240 prop "mapM (+1)" $
241 transform (fmap (+1)) (Async.mapM (\x -> return (x + 1)))
243 -- XXX Need to use eq instead of sortEq for ahead oeprations
244 -- Binary append
245 prop1 "append [] []"
246 $ cmp (Async.parLazy [Stream.nil, Stream.nil]) sortEq []
247 prop1 "append [] [1]"
248 $ cmp (Async.parLazy [Stream.nil, Stream.fromPure 1]) sortEq [1]
249 prop1 "append [1] []"
250 $ cmp (Async.parLazy [Stream.fromPure 1, Stream.nil]) sortEq [1]
251 prop1 "append [0] [1]"
252 $ let stream = Async.parLazy [Stream.fromPure 0, Stream.fromPure 1]
253 in cmp stream sortEq [0, 1]
255 prop1 "append [0] [] [1]"
256 $ let stream =
257 Async.parLazy
258 [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]
259 in cmp stream sortEq [0, 1]
261 prop1 "append2 left associated"
262 $ let stream =
263 Stream.fromPure 0
264 `Async.append2` Stream.fromPure 1
265 `Async.append2` Stream.fromPure 2
266 `Async.append2` 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.append2` (Stream.fromPure 1
273 `Async.append2` (Stream.fromPure 2
274 `Async.append2` 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.append2` Stream.fromPure y
279 leaf11 = leaf 0 1 `Async.append2` leaf 2 (3 :: Int)
280 leaf12 = leaf 4 5 `Async.append2` leaf 6 7
281 stream = leaf11 `Async.append2` leaf12
282 in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
284 prop1 "combineWith (maxThreads 1)"
285 $ let stream =
286 Async.combineWith (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 (Stream.fromPure (,)) (Stream.fromPure 1 `Async.append2` Stream.fromPure 2)
293 s2 = Async.apply s1 (Stream.fromPure 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 = Stream.fromPure (1,)
299 s2 = Async.apply s1 (Stream.fromPure 2 `Async.append2` Stream.fromPure 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.parLazy
319#endif
320 describe "Exception propagation" $ exceptionPropagation Async.append2
321 -- Ad-hoc tests
322 it "takes n from stream of streams" $ takeCombined 2