git.iko.soy/streamly
on 430f06d help sign in
streamly/test/Streamly/Test/Data/Stream/Concurrent.hs
319 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.parList id [constr b, constr c])
72 let s2 = Async.parList id [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.fromList [1..10])
135 `f` Stream.fromEffect (throwM (ExampleException "E"))
136 `f` (Stream.fromList [1..10])
137 try (tl (Stream.nil `f` nested `f` (Stream.fromList [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.parSequence id stream))
144 `shouldReturn`
145 (Left (ExampleException "E") :: Either ExampleException [Int])
147 it "concatMap throwM" $ do
148 let s1 = Async.parList id $ fmap Stream.fromPure [1..4]
149 s2 = Async.parList id $ fmap Stream.fromPure [5..8]
150 try $ tl (
151 let bind = flip (Async.parConcatMap id)
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) -> Spec
171timeOrdering f = do
172 it "Parallel event ordering check" $
173 Stream.fold Fold.toList (f [event 4, event 3, event 2, event 1])
174 `shouldReturn` [1..4]
176 where event n = Stream.fromEffect (threadDelay (n * 200000) >> return n)
177#endif
179-------------------------------------------------------------------------------
180-- Some ad-hoc tests that failed at times
181-------------------------------------------------------------------------------
183takeCombined :: Int -> IO ()
184takeCombined n = do
185 let constr = Stream.fromList
186 let s = Async.parList id [constr ([] :: [Int]), constr ([] :: [Int])]
187 r <- Stream.fold Fold.toList $ Stream.take n s
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.parSequence 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.parEval id . fmap (+1))
234 asyncSpec $ prop "parSequence" . sequenceReplicate
235 -- XXX Need to use asyncSpec in all tests
236 prop "mapM (+1)" $
237 transform (fmap (+1)) (Async.parMapM id (\x -> return (x + 1)))
239 -- XXX Need to use eq instead of sortEq for ahead oeprations
240 -- Binary append
241 prop1 "append [] []"
242 $ cmp (Async.parList id [Stream.nil, Stream.nil]) sortEq []
243 prop1 "append [] [1]"
244 $ cmp (Async.parList id [Stream.nil, Stream.fromPure 1]) sortEq [1]
245 prop1 "append [1] []"
246 $ cmp (Async.parList id [Stream.fromPure 1, Stream.nil]) sortEq [1]
247 prop1 "append [0] [1]"
248 $ let stream = Async.parList id [Stream.fromPure 0, Stream.fromPure 1]
249 in cmp stream sortEq [0, 1]
251 prop1 "append [0] [] [1]"
252 $ let stream =
253 Async.parList id
254 [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]
255 in cmp stream sortEq [0, 1]
257 let async = Async.parTwo id
258 prop1 "append2 left associated"
259 $ let stream =
260 Stream.fromPure 0
261 `async` Stream.fromPure 1
262 `async` Stream.fromPure 2
263 `async` 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` (Stream.fromPure 1
270 `async` (Stream.fromPure 2
271 `async` 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` Stream.fromPure y
276 leaf11 = leaf 0 1 `async` leaf 2 (3 :: Int)
277 leaf12 = leaf 4 5 `async` leaf 6 7
278 stream = leaf11 `async` leaf12
279 in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]
281 prop1 "combineWith (maxThreads 1)"
282 $ let stream =
283 Async.parTwo (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.parApply id (Stream.fromPure (,)) (Stream.fromPure 1 `async` Stream.fromPure 2)
290 s2 = Async.parApply id s1 (Stream.fromPure 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 = Stream.fromPure (1,)
296 s2 = Async.parApply id s1 (Stream.fromPure 2 `async` Stream.fromPure 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.parConcat id
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.parConcatMap id (const (Stream.fromList [1..n])))
314#ifdef DEVBUILD
315 describe "Time ordering" $ timeOrdering Async.parList id
316#endif
317 describe "Exception propagation" $ exceptionPropagation async
318 -- Ad-hoc tests
319 it "takes n from stream of streams" $ takeCombined 2